Compare commits

..

10 Commits

Author SHA1 Message Date
Gud Boi 749ca1f09c Add `piker-fsp-expert` skill
Document FSP lifecycle, SHM publication, backfill ordering,
causal signal-processing semantics, and deterministic tests.
Record research directions separately from implementation and
expose the package through the relative Claude skill symlink.

(this commit msg was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 13:04:17 -04:00
Gud Boi 224399147d Honor M4 renderer cache invalidation
Preserve the renderer's `reset_cache` result when publishing
rebuilt paths so zoom changes invalidate Qt's cached graphics.
Retain the earlier live-debug provenance for cadence, negative
volume, repaint suppression, and downsample cache repair.

Prompt-IO: ai/prompt-io/opencode/20260903T232525Z_fadab3d2_prompt_io.md

(this commit msg was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 12:49:53 -04:00
Gud Boi 5b4622fb39 Skip unchanged realtime FSP redraws
Cache destination bounds and timestamp/value bytes on `Viz`.
Skip identical quote-driven refreshes, including stable NaNs,
while changed samples and forced history repairs still redraw.
Add a regression covering each invalidation path.

(this commit msg was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 12:49:42 -04:00
Gud Boi 20406210b3 Clamp negative provider volume before FSP updates
Clamp negative trade sizes and cumulative bar volume in sampling,
and sanitize history and realtime volume in `tina_vwap()` and
`dolla_vlm()`. Provider resets must not create negative outliers
that hide the derived curve. Cover the history and tick paths.

(this commit msg was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 12:47:07 -04:00
Gud Boi 07d8cf5a3d Fix FSP history and sample synchronization
Bootstrap FSP output and timestamps from one source snapshot,
validate the history schema, and compare absolute SHM bounds.
Subscribe before bootstrap, replay history revisions, filter
foreign sampler notices, and append each destination step once.
Close replaced generators before signaling completion.

Use declared `Flume` sample periods so market-closure gaps cannot
be mistaken for the realtime cadence. Cover bounds, snapshot
isolation, backfill routing, sample steps, and targeted redraws.

(this commit msg was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 12:46:20 -04:00
Gud Boi d5b20ce350 Formalize `FspStream` and fix builtin call contracts
Export `FeedQuote` and `Tick`, describe historical and realtime
FSP yields, and preserve callable parameters through `@fsp`.
Align builtin history arrays and emit named realtime updates.

Keep the staged Tuicr review fixes. Let Numba specialize `ema()`
lazily so omitted defaults work alongside explicit arguments,
retaining nopython/nogil compilation and single-sample seeding.
Cover builtin yields and EMA argument forms with regressions.

Prompt-IO: ai/prompt-io/opencode/20260908T020619Z_fadab3d2_prompt_io.md
Prompt-IO: ai/prompt-io/codex/20260915T155229Z_fadab3d2_prompt_io.md

(this patch was generated in some part by `codex` using
`gpt-6` (`openai`))
2026-09-15 12:17:02 -04:00
Gud Boi fadab3d2bc Target FSP history graphics by SHM token
Replace whole-linked-chart refreshes after FSP history rebuilds
with updates selected by normalized destination `NDToken` identity.

Clear each matching `Viz` formatter and max/min cache, redraw
its path, then rescale only the affected view. This keeps primary
and foreign market curves stable while near-term history is
backfilled.

Record the concrete piker `PlotItem`, `ChartView`, `Axis`, and
`YAxisLabel` relationships used by the redraw path.

Prompt-IO: ai/prompt-io/opencode/20260828T195032Z_9115d24d_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-09-01 22:07:22 -04:00
Gud Boi 9115d24da0 Fix string errors in `.ib.broker` relay
Normalize structured and raw IB errors without indexing strings as
message dictionaries.

Deats,
- match each error form before request-id handling;
- parse `[code N]` and `[Errno N]` prefixes without crashing;
- cover raw errors through `deliver_trade_events()`.

Prompt-IO: ai/prompt-io/opencode/20260828T063126Z_1606122b_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-28 02:52:12 -04:00
Gud Boi 1606122b45 Add `piker-clearing-expert` skill
Capture EMS/order-control ownership, identifier boundaries, ledger
and position invariants, backend variation, and current known hazards.

Store the canonical bundle under `.agents/skills/` and bridge Claude
discovery through a portable relative link.

Also,
- add architecture, debugging, and test-map references;
- document the provider-neutral source convention;
- ignore the canonical `wkts/` runtime root.

Prompt-IO: ai/prompt-io/opencode/20260828T021411Z_6ca6ab2a_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-27 23:24:54 -04:00
Gud Boi 7b51927f6b Fix `SymbologyCache.search()` key matching
RapidFuzz treated `mktmaps` values as fuzzy choices and called
`len()` on `MktPair` structs during paper-ledger persistence.

Search string keys explicitly and resolve each match back to its
struct so native market-ID aliases survive legacy FQME
qualification.

Also,
- cover alias preservation and the paper-ledger TOML round trip.

Prompt-IO: ai/prompt-io/opencode/20260828T013057Z_6ca6ab2a_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-27 23:11:43 -04:00
48 changed files with 3477 additions and 397 deletions

View File

@ -0,0 +1,174 @@
---
name: piker-clearing-expert
description: >
Piker clearing, EMS/order-control, accounting, position, and
trade-ledger expertise. Apply when changing or debugging Order,
Status, Brokerd* protocols, live or paper fills, dark orders,
order IDs, TransactionLedger persistence, Account/Position/PPU,
FQME/bs_mktid identity, SymbologyCache lookups, or broker trade
normalization.
compatibility: >
Requires a piker checkout and familiarity with Python, trio,
tractor, msgspec Structs, and backend-specific trading APIs.
metadata:
author: goodboy
version: "1.0"
---
# Piker Clearing Expert
Use this mental model before touching order control or accounting:
```text
OrderClient / UI
-> emsd routes intent and owns order-dialog state
-> live brokerd OR paperboi executes and reports events
-> executing backend owns fill-to-accounting integration
-> emsd translates and broadcasts broker events
-> UI consumes order state and position summaries
```
The EMS is not the trade ledger. It correlates IDs, runs dark
predicates, and relays lifecycle events. Paper and accounting-aware
live backends may normalize fills into transactions, update local
ledger/account state, and emit `BrokerdPosition` summaries. Inspect
each live backend; this integration is not uniform.
## Runtime ownership
- `OrderClient` owns client intent and local sent-order tracking.
- `emsd` owns `Router`, `DarkBook`, active `Status` dialogs,
subscribers, and `oid <-> reqid` correlation.
- `brokerd.<broker>` owns credentialed live order control.
- `paperboi.<broker>` owns simulated execution and paper accounting.
- `TransactionLedger` owns trade-record persistence, not order state.
- `Account` owns positions keyed by backend-system market identity.
- UI `Position.update_from_msg()` is a summary reset, not durable
transaction accounting.
For daemon placement, context streams, cancellation, and actor-local
state, also load `piker-conc-expert`.
## Event lifecycle
1. A client sends `Order` or `Cancel` through `OrderClient`.
2. `process_client_order_cmds()` creates or updates EMS `Status`.
3. EMS routes a `BrokerdOrder` or stores a dark trigger predicate.
4. The executing backend acknowledges with `BrokerdOrderAck`.
5. EMS records the current `oid <-> reqid` relation.
6. Accepted, ack-correlated status/fill events become client updates.
7. The backend integrates fills into accounting when supported.
8. A resulting `BrokerdPosition` is relayed through EMS to clients.
Cancellation may arrive before acknowledgement. Preserve
`Status.cancel_called`; once the ack supplies `reqid`, EMS can issue
the deferred `BrokerdCancel`.
## Identifier namespaces
Never substitute one namespace merely because two values happen to
match in a backend:
| ID | Owner and invariant |
|---|---|
| `oid` | Stable client/EMS order-dialog ID across submit, modify, cancel |
| `reqid` | Backend order-control ID established by an ack; it may change |
| venue order ID | Native API identity; may differ from both IDs above |
| `tid` | Stable clear ID used to deduplicate position events and TOML keys |
| `fqme` | Piker-normalized market address and symcache lookup key |
| `bs_fqme` | FQME without its final broker suffix |
| `bs_mktid` | Backend-system market key and `Account.pps` key |
Normalize every durably persisted `tid` to a stable string. Although
`Transaction.tid` permits integers, TOML mapping keys do not.
Paper transactions commonly set `Transaction.bs_mktid` to the full
FQME. The embedded `Position.mkt.bs_mktid` can still retain the
provider-native market ID. Check the concrete object and table owner
before asserting key equality.
## Accounting invariants
- `Transaction.size` is signed; preserve side semantics during
broker normalization.
- `Transaction` is the normalized in-memory interchange. Live ledger
records may remain backend-native and normalize through
`mod.norm_trade()` when read.
- `Position._events` is keyed by `tid`; `add_clear()` is the
idempotence boundary.
- `Position.ppu` and `Position.cumsize` are derived from transaction
events, not independent mutable truth.
- Normalized accounting transactions need a stable, non-null
`bs_mktid`; `Account.pps` uses it as the position key.
- `Account.update_from_ledger()` resolves market metadata from
`SymbologyCache.mktmaps` or its explicit fallback table.
- UI position messages are projections. Do not feed their synthetic
reset event back into durable broker accounting.
## Persistence contracts
Trade ledgers live under the accounting ledger directory as
`trades_<broker>_<account>.toml`. Top-level keys are transaction IDs;
record schemas are backend-specific unless the paper engine wrote a
normalized transaction dictionary.
`open_trade_ledger()` writes on context exit when its data comparison
detects a change or `rewrite=True`. Its snapshot is shallow, so nested
in-place record mutation can evade dirty detection. Explicit writes can
also fail inside fill handlers. Treat all persistence points as part of
the live execution path.
Accounts persist active positions. `dump_active()` separates open and
net-zero positions. Do not assume `minimized_clears()` safely preserves
size/PPU: current sign-transition and longer net-zero slicing lacks
round-trip coverage and can retain the wrong event subset.
## Market identity and symcache
`SymbologyCache` has three distinct key-addressed mappings:
- `assets`: provider asset ID to `Asset`;
- `pairs`: native `bs_mktid` to backend pair `Struct`;
- `mktmaps`: searchable FQME or explicit alias to normalized
`MktPair`.
Search mapping keys, then resolve values. RapidFuzz treats mapping
values as choices; passing `mktmaps` directly makes it operate on
`MktPair` objects. Preserve in-memory aliases because IB can store both
native IDs and FQMEs for one `MktPair`. Do not persist such aliases
without fixing reload: `from_dict()` currently requires every key to
equal `mkt.fqme`.
## Change workflow
Before editing:
1. Identify the owning layer: client, EMS, executor, accounting, or UI.
2. Write down every ID/key transition across that boundary.
3. Trace both normal and cancel-before-ack event ordering.
4. Determine whether records are native, normalized, or summaries.
5. Check actor-local caches and context-exit writes.
6. Add a deterministic regression at the narrowest broken contract.
7. Run live/backend tests only when their credentials and side effects
are explicitly authorized.
Do not fix an accounting defect by adding state to EMS, or fix an EMS
correlation defect by rewriting durable transaction history.
## Canonical source map
- `piker/clearing/_messages.py`: client and brokerd protocol types.
- `piker/clearing/_client.py`: `OrderClient` and `open_ems()`.
- `piker/clearing/_ems.py`: router, dark book, ID mapping, translation.
- `piker/clearing/_paper_engine.py`: simulated execution/accounting.
- `piker/accounting/_ledger.py`: transactions and ledger persistence.
- `piker/accounting/_pos.py`: positions, accounts, PPU event state.
- `piker/accounting/_mktinfo.py`: `Asset`, `MktPair`, FQME schema.
- `piker/data/_symcache.py`: provider symbology mappings and search.
- `piker/accounting/calc.py`: transaction ordering and PPU/PnL math.
- `piker/brokers/*/broker.py`: backend order-control integration.
- `piker/brokers/*/ledger.py`: backend trade normalization.
See [architecture.md](architecture.md) for complete flows,
[gotchas.md](gotchas.md) for symptom-driven diagnosis, and
[test-map.md](test-map.md) for verification choices.

View File

@ -0,0 +1,210 @@
# Clearing And Accounting Architecture
## Protocol direction
Client-facing messages and broker-facing messages intentionally differ.
```text
client -> EMS
Order, Cancel
EMS -> executor
BrokerdOrder, BrokerdCancel
executor -> EMS
BrokerdOrderAck, BrokerdStatus, BrokerdFill,
BrokerdError, BrokerdPosition
EMS -> subscribed clients
Status, BrokerdPosition
```
`Order` carries the client `oid`, account, market, price, size, action,
and execution mode. `BrokerdOrderAck` establishes the executor's
`reqid`; subsequent broker messages should carry that ID so EMS can
resolve the client dialog.
`Status` is mutable current dialog state, not a history envelope. Exact
client states include `pending`, `open`, `dark_open`, `triggered`,
`fill`, `closed`, `canceled`, and `error`. Audit existing translation
before adding a state; backend `BrokerdStatus` vocabulary is narrower
and current producers are not fully conformant.
## EMS routing
`_emsd_main()` serves each client/market context. The actor-local
`Router` owns books, subscribers, order dialogs, and cached live trades
relays. Paper mode opens per-symbol relay contexts under the same broker
key, a current overwrite/pop hazard for concurrent paper symbols.
`open_brokerd_dialog()` is the session live-versus-paper selection
point:
- live execution talks to `brokerd.<broker>`;
- paper execution talks to `paperboi.<broker>`;
- chart-only and paper sessions should not boot credentialed brokerd.
Do not infer paper execution from each order's `exec_mode`. Session
mode selects the executor; per-order mode primarily distinguishes dark
from immediately routed orders, despite incomplete literal annotations.
`process_client_order_cmds()` handles submit, modify, cancel, dark,
and alert intent. It updates EMS dialog state before routing broker
commands. `translate_and_relay_brokerd_events()` applies acks, statuses,
fills, errors, and positions to dialogs and subscribers.
The `DarkBook` stores local predicates and watches quote ticks. When a
predicate triggers, `clear_dark_triggers()` emits the live broker order
and updates the existing client dialog rather than inventing a second
order identity.
## Ack and cancel ordering
Required actionable ordering is submit, ack, open, zero or more fills,
then terminal status. EMS handles cancel-before-ack, but it does not
buffer arbitrary status-before-ack events; those currently fall through
as unhandled. Backends must ack before actionable statuses.
The important race is:
```text
client Cancel(oid)
-> EMS has no reqid yet
-> mark Status.cancel_called
BrokerdOrderAck(oid, reqid)
-> install mapping
-> send deferred BrokerdCancel(reqid)
```
Do not drop a pre-ack cancel and do not send a broker cancel with an
`oid` unless the backend explicitly uses that value as its `reqid`.
## Paper execution
`PaperBoi` maintains simulated orders, opens the market feed, and fills
orders when L1 ask/bid quotes cross their prices. `trade`/`last` ticks
zip buy and sell iterators, so a lone resting side currently may not
fill from those tick types. `fake_fill()` is the core fill-to-account
path:
1. emit `BrokerdFill`;
2. emit terminal `BrokerdStatus` for the simulated order;
3. construct a signed normalized `Transaction`;
4. `TransactionLedger.update_from_t()`;
5. `Account.update_from_ledger()`;
6. persist the ledger and account;
7. emit `BrokerdPosition` with current size and PPU.
The protocol notifications and persistence occur in one task but are
not a database transaction. If persistence fails after fill/status
publication, restart/reconciliation behavior matters. Do not reorder
the sequence casually.
Partial fills currently reuse `oid` as `tid`. A later partial therefore
overwrites the paper ledger record and is rejected by position-event
dedupe. Fix unique fill IDs before relying on paper partial-fill math.
`open_trade_dialog()` opens symcache, paper ledger, account, and feed
resources under structured concurrency. An exception from any nested
resource, including ledger context exit, cancels the dialog and its feed
stream.
## Live backend variation
There is no universal live-broker accounting implementation:
- IB normalizes execution/commission events, updates ledger/account
state, and emits position updates.
- Kraken has backend trade normalization and position reconstruction,
but its live path writes account state without inserting every new
trade into `TransactionLedger`.
- Binance currently relays venue order/position events and does not
share every local `TransactionLedger`/`Account` path.
Inspect the backend's `broker.py`, `ledger.py`, and `norm_trade()` before
moving generic logic. A paper behavior is not automatically a live
backend contract.
## Ledger normalization
`TransactionLedger.data` is the durable record mapping. For paper it
contains transaction-like dictionaries. For live brokers it can contain
raw provider records.
`iter_txns()` sorts records and chooses normalization:
- paper uses `_paper_engine.norm_trade()`;
- live accounts use `brokermod.norm_trade()`.
Normalizer output should carry datetime, signed size, price, cost,
FQME, backend market ID, and transaction ID. Some normalizers return
`None` for unusable records; `iter_txns()` does not filter it, while
`to_txns()` does. Audit direct iterator consumers.
`write_config()` sorts records for stable persistence. Paper rewrite
logic migrates old symbol fields and qualifies legacy FQMEs through the
symcache. Search can return no matches; callers which select the first
result need an explicit fallback or a descriptive failure policy.
## Position event model
`Position` holds a market and a transaction-event mapping. New clears
are deduplicated by `tid`. Current size and PPU are recomputed from the
event sequence.
`Account.update_from_ledger()`:
1. receives normalized transactions;
2. resolves each market through `mktmaps` or a fallback table;
3. looks up or creates the position by `Transaction.bs_mktid`;
4. adds unseen clears;
5. returns visited positions, including duplicates whose clear was
rejected by `add_clear()`.
`minimized_clears()` intends to discard irrelevant history after
net-zero or sign transitions, but current slicing is not proven safe
and can reconstruct the wrong active size/PPU. Treat it as a known bug
surface until sign-transition round trips are covered.
`Position.update_from_msg()` serves UI summaries by replacing local
event state with a synthetic clear representing reported size and PPU.
That projection is intentionally lossy.
## Symbology contracts
`MktPair.fqme` is a normalized endpoint address assembled from pair,
venue, expiry/contract information, and broker. `MktPair.bs_fqme`
drops the broker suffix. `MktPair.bs_mktid` is the provider-native
market identity.
`SymbologyCache.load()` attempts to obtain provider assets and pair
structs, calls `get_mkt_info()`, and populates conventional mappings.
Assets are optional; missing/empty pair support returns early.
```text
assets[provider_asset_id] -> Asset
pairs[pair.bs_mktid] -> backend Pair Struct
mktmaps[mkt.fqme] -> MktPair
```
Some backends add aliases such as `mktmaps[bs_mktid]`. Search must
return the matched mapping key rather than rebuilding a key from the
value, otherwise aliases silently collapse. Current serialization then
writes aliases which reload rejects because `from_dict()` asserts each
key equals `mkt.fqme`.
## Persistence boundaries
`open_trade_ledger()` loads account TOML and compares a shallow snapshot
in `finally`. Top-level additions/replacements trigger writes; nested
in-place mutation may not. `open_account()` similarly mediates account
state. These contexts run inside actor tasks, so serialization errors
are runtime failures, not merely offline maintenance failures.
When changing schemas:
- preserve backend-native records unless migration is explicit;
- keep old paper field migration deterministic;
- avoid fuzzy qualification when an exact `bs_mktid`/FQME exists;
- test context exit, not only in-memory updates;
- inspect emitted positions after disk round trips.

View File

@ -0,0 +1,146 @@
# Clearing And Ledger Gotchas
## Trade dialog dies immediately after a fill
Likely cause: an explicit ledger/account write in the fill handler, or
a later context-exit write, raised after event publication.
Inspect, in order:
1. the first non-cancellation exception in the actor traceback;
2. `PaperBoi.fake_fill()` or the live backend fill handler;
3. explicit and context-exit `write_config()` calls;
4. FQME/`bs_mktid` qualification and TOML-compatible values;
5. the persisted ledger/account files for a partial update.
Tractor cancellation logs are often the consequence. Root-cause the
inner serialization, lookup, or normalization exception first.
## `MktPair` has no `len()` in RapidFuzz
Cause: a mapping of string keys to `MktPair` values was passed directly
to `rapidfuzz.process.extract()`. Mapping values become fuzzy choices.
Fix: search a list/view of string keys and resolve matched keys back
through the original mapping. Do not add sequence methods to `MktPair`
and do not use `processor=str` when aliases must be preserved.
## Fuzzy qualification raises `IndexError`
Cause: the caller selected the first match from an empty result after
the score cutoff rejected every key.
Decide explicitly whether to retain the input identity, reject the
record with a descriptive exception, or obtain missing market metadata.
Do not lower the cutoff blindly; the wrong market corrupts accounting.
## Cancel is ignored or sent with the wrong ID
Cause: confusion between client `oid` and backend `reqid`, commonly
during cancel-before-ack ordering.
Verify the ack installed both mapping directions and that
`Status.cancel_called` survives until `reqid` exists. Log both IDs and
the actor/backend account at every transition.
## Fills appear but positions do not move
Check these boundaries:
- the backend emitted `BrokerdFill` but never normalized a transaction;
- transaction size lost its buy/sell sign;
- `tid` collided with an existing position event;
- `Transaction.bs_mktid` selected a different `Account.pps` key;
- `mktmaps` lacked the exact transaction identity;
- position publication happened before accounting update;
- the backend intentionally reports venue positions instead of local
ledger-derived positions.
## Duplicate or inflated position size
Do not deduplicate by price/time alone. Confirm stable transaction IDs,
chronological sorting, and `Position.add_clear()` idempotence. Broker
execution IDs and order IDs are not interchangeable.
For partial fills, several `tid`s may correctly share one order
`reqid`. Collapsing them loses clears; replaying one `tid` twice
inflates nothing only if the dedupe boundary is preserved.
Current paper fills use `tid=oid`, so later partials overwrite the
ledger record and are deduplicated out of the position. Treat paper
partial fills as broken until each clear receives a stable unique ID.
## Paper and live state disagree
Paper uses normalized local transactions and can use the full FQME as
its account position key. A live backend may use native IDs, raw ledger
records, or venue-reported positions.
Compare normalized `Transaction`s and emitted `BrokerdPosition`s, not
raw TOML dictionaries across backends.
## Account appears under the wrong name
Separate UI/EMS account aliases from backend account IDs and ledger file
names. Backends may prefix account names for routing while persistence
uses an unprefixed native account key.
Trace account identity through `open_ems()`, broker dialog startup,
ledger opening, and `BrokerdPosition.account` before changing naming.
## A module-global cache is unexpectedly empty
Symcache, clients, contracts, and order tables are actor-local. Data
loaded in `datad` is not automatically available in `brokerd` or
`paperboi`. Also, `_symcache._caches` is currently read but never
populated, so do not assume repeated `open_symcache()` calls hit it.
Load `piker-conc-expert` and audit actor-local globals, async caches,
and dialog startup warming. Never rely on an import side effect from a
sibling actor.
## Teardown emits a cancellation storm
Find the first application exception. Structured cancellation then
closes feed streams, broker streams, and nested contexts. Avoid masking
the originating exception with broad cancellation handling.
For stream ownership and context cancellation semantics, use
`piker-conc-expert`.
## Broker status is logged as unhandled
EMS currently accepts only ack-correlated `open`, `closed`, and
`canceled` broker statuses. Declared `pending` and backend-produced
`fill`/`filled` variants fall through, as do statuses arriving before
ack. Verify producer vocabulary and ack ordering before adding another
consumer branch.
## Another client sees order or position events
Current translation broadcasts order updates by FQME and positions to
all attached clients; `Router.dialogs` is not used as an event privacy
filter. Treat subscriber isolation as incomplete, not guaranteed.
## Debug breadcrumb
Capture one row per event with:
```text
actor, backend, account, fqme, bs_mktid,
oid, reqid, venue_order_id, tid,
message_type, status, action, signed_size, price, timestamp
```
Then compare:
1. client command order;
2. EMS status history;
3. broker event order;
4. normalized transactions;
5. position event IDs and computed size/PPU;
6. ledger/account disk state;
7. emitted position summaries.
This separates protocol loss, identity skew, accounting math, and
persistence failure without guessing from UI state.

View File

@ -0,0 +1,84 @@
# Clearing And Accounting Test Map
## Deterministic first-pass tests
Use the narrowest test that owns the broken contract:
| Area | First target |
|---|---|
| Ledger/account persistence | targeted node in `tests/test_accounting.py` |
| EMS bad-backend handling | `tests/test_ems.py::test_ems_err_on_bad_broker` |
| IB request/response routing | `tests/test_ib_method_proxy.py` |
| IB history normalization | `tests/test_ib_history.py` |
| Service/actor startup | targeted node in `tests/test_services.py` |
Known missing high-value regressions include paper partial-fill IDs,
status-before-ack handling, `minimized_clears()` sign-transition round
trips, nested ledger dirty detection, and symcache alias reload.
For `SymbologyCache.search()` and paper ledger qualification, cover:
- real `Asset` and `MktPair` values;
- a native-ID alias and canonical FQME key;
- the actual `TransactionLedger.write_config()` path;
- persisted TOML fields after the round trip;
- no use of live APIs or user configuration.
## Side-effectful tests
Do not run these without explicit authorization and the repository test
harness guidance:
- paper EMS tests which open live market/symbology feeds;
- live Binance/Kraken feed suites;
- IB account tests using configured accounts;
- tests which write tracked ledger/account fixtures;
- broker dialogs requiring credentials or venue connectivity.
After any accounting test using `tests/_inputs`, inspect its diff. Some
context managers can rewrite fixture ledgers on exit.
## Regression design
Prefer a direct contract regression over a full actor test when the
failure is synchronous and deterministic. Use an actor-level test when
the bug depends on:
- ack/cancel or fill/status interleaving;
- context cancellation or stream ownership;
- actor-local state/cache separation;
- daemon selection or startup;
- publication ordering across EMS and broker streams.
Every regression should document the original failure, triggering
state, violated invariant, arrangement, and proof of the fix.
## Verification layers
1. `git diff --cached --check` for the exact staged boundary.
2. Ruff with the repository's intended rule profile.
3. A deterministic unit/filesystem regression.
4. Adjacent accounting or EMS tests which do not require live state.
5. Explicitly authorized backend/actor integration tests.
Do not interpret a passing ellipsis-body or hard-skipped test as
behavioral coverage. Read the selected node before using it as a gate.
## Manual reconciliation check
For production-only failures, compare one fill end to end:
```text
BrokerdFill
-> normalized Transaction
-> ledger record
-> Position._events[tid]
-> Position.cumsize / ppu
-> account TOML
-> BrokerdPosition
-> UI summary
```
All identities, sign, price, and account values should remain explainable
at each arrow. A mismatch identifies the owning boundary for the next
focused test.

View File

@ -0,0 +1,78 @@
---
name: piker-fsp-expert
description: Piker financial signal processing expertise. Apply when changing or debugging FSP definitions, Cascade execution, derived shared-memory arrays, history/realtime handoff, backfill synchronization, FSP graph composition, online financial algorithms, or FSP chart lifecycle and scaling.
---
# Piker FSP Expertise
Treat financial signal processing as causal, revision-aware processing of
event-time market data, not as generic batch array manipulation.
## Start Here
1. Identify the source `Flume`, timeframe, event clock, and writer actor.
2. State whether the operation is historical bootstrap, backfill revision,
current-sample mutation, or sample-step append.
3. Record source and destination absolute SHM bounds before reasoning from
lengths or `ShmArray.index`.
4. Separate compute, SHM publication, IPC wakeup, and graphics costs before
optimizing.
5. Preserve causality, warm-up behavior, missing-data semantics, and output
schema in every implementation and test.
Read [architecture.md](architecture.md) before modifying runtime code. Read
[signal-processing.md](signal-processing.md) before adding an algorithm. Use
[roadmap-and-research.md](roadmap-and-research.md) for design and dependency
work, and [test-map.md](test-map.md) for verification.
## Core Contracts
- `piker.data` owns provider ingestion, source `Flume`s, history publication,
quote fan-out, OHLC mutation, and `samplerd` step events.
- `piker.fsp` owns operator declarations, historical computation, realtime
state updates, and destination SHM publication.
- `piker.ui._fsp.FspAdmin` currently owns FSP worker selection, destination
allocation, cascade lifetime, and chart attachment.
- One logical writer owns each destination FSP array.
- Output event time and absolute indices derive from the source series, not
task scheduling or quote arrival order.
- A source prepend changes historical provenance even when the current sample
index is unchanged. Recompute or patch the affected destination range.
- One-step destination lag is normal between source sampling and destination
append. A destination lead or unexplained bound shift is not.
- Backfill notifications are source and timeframe specific even though the
current sampler transport broadcasts them globally.
- Only graphics backed by a rebuilt destination should refresh. Never use an
FSP update as an unconditional whole-chart redraw request.
## Editing Workflow
1. Trace the operator from `@fsp` declaration through `cascade()`,
`connect_streams()`, destination SHM, `FspAdmin`, and every consuming
`Viz`.
2. Write an event-order table covering source writes, bound publication,
notifications, cancellation, recomputation, and destination publication.
3. Add deterministic tests for history and realtime phases independently.
4. Add a controlled interleaving test for every race fix.
5. Profile representative visible ranges and feed rates before adding a
dependency or native boundary.
## Companion Skills
- Use `piker-conc-expert` for Tractor contexts, actor topology, cancellation,
service ownership, and cross-actor state.
- Use `timeseries-optimization` for NumPy/Polars lookup and vectorization.
- Use `piker-profiling` for distributed performance measurements.
- Use `pyqtgraph-optimization` for renderer and graphics batching work.
- Use `piker-clearing-expert` when a derived signal reaches EMS or accounting.
## Do Not
- Do not hide a historical mismatch merely to stop a resync log.
- Do not infer a complete synchronization state from array length alone.
- Do not run Python loops over historical numeric arrays without evidence.
- Do not apply regular-grid DSP blindly to irregular tick arrival times.
- Do not introduce an external stream runtime that duplicates Tractor before
measuring an isolated protocol or kernel prototype.
- Do not route strategy signals directly around EMS order-state, risk, or
accounting contracts.

View File

@ -0,0 +1,117 @@
# FSP Runtime Architecture
## Current Topology
```text
provider backend
-> datad feed bus
-> source Flume
-> rt ShmArray (typically 1 second)
-> hist ShmArray (typically 60 seconds)
-> samplerd sample and backfill events
chart actor
-> FspAdmin
-> chart.fsp_N worker actor
-> cascade()
-> connect_streams()
-> destination Flume / ShmArray
-> targeted Viz updates
```
The source path is implemented mainly by `piker.data.feed`,
`piker.data.flows`, `piker.data._sampling`, and `piker.tsp._history`. The FSP
path is implemented by `piker.fsp._api`, `piker.fsp._engine`, and operator
modules such as `_volume` and `_momo`. Worker and graphics lifecycle currently
live in `piker.ui._fsp` and `piker.ui._display`.
## Operator Contract
`@fsp` wraps an async generator. Its first yield is historical output computed
from the source array. Later yields are `(field, value)` realtime mutations.
`connect_streams()` converts the first yield into the destination structured
array, aligns time and absolute bounds, then updates the current destination
row from realtime quote events.
`samplerd` owns row advancement. At a new source sample, the cascade copies or
zeros the previous destination row and aligns destination timestamps to the
source. Quote-time mutations and sample-time appends are distinct operations.
## SHM Synchronization
Treat each array as an absolute half-open range:
```text
[first, last)
```
For a normal source history prepend:
```text
before: src=[10000, 12000) dst=[10000, 12000)
after: src=[ 8000, 12000) dst=[10000, 12000)
```
The current step is unchanged, but destination history is missing 2,000 rows.
The destination must be recomputed or incrementally repaired.
For a normal one-step source lead:
```text
src=[8000, 12001) dst=[8000, 12000)
```
The cascade should append exactly one destination row, not recompute history.
Use `_first.value` and `_last.value` for synchronization. `ShmArray.index` is
the last bound modulo capacity and is not a complete absolute position.
Snapshot bounds once per check; shared first and last counters do not form a
transactional pair.
## Backfill Event Ordering
1. `start_backfill()` receives an older provider frame.
2. `shm_push_in_between()` writes the frame and publishes the earlier first
bound.
3. `notify_backfill()` asks `samplerd` to broadcast the market and timeframe.
4. Relevant cascades treat the event as a history revision. A prepend changes
bounds; an in-place gap repair may not.
5. Each cascade cancels its current compute task and waits for completion.
6. The historical phase recomputes against the newest readable source range.
7. Destination bounds are published and the UI receives an `fsp_update`.
8. Only `Viz` objects backed by that destination token redraw.
Multiple provider frames may arrive while a recomputation runs. The cascade
must reject mixed-bound bootstrap results, replay in-place revisions, recheck
bounds after restart, and converge on the newest source range.
## Actor-Local State
`Fsp._flow_registry`, attached SHM handles, caches, and nurseries are local to
each actor. Never assume worker siblings share warmed registries or Python
objects. Graph dependencies currently rely on startup order, as demonstrated
by `flow_rates` starting only after `dolla_vlm`.
## Cross-Subsystem Boundaries
- Data ingest should normalize identity and event-time semantics before an FSP
consumes a stream.
- UI renderers should consume revision/range information without controlling
computation correctness.
- Clearing may consume a derived signal, but EMS remains responsible for order
intent, status, routing, fills, positions, and accounting.
- FSP output exposed as a feed needs the same fan-out, identity, lifecycle,
and backpressure contracts as provider feeds.
## Architectural Gaps
The current protocol infers history revisions from shared bounds. A future
revision message should carry at least:
```text
(fqme, timeframe, revision, first, last, start_ts, end_ts)
```
Destination publication should identify the source revision used. This allows
coalescing, range repair, stale-result rejection, and deterministic graph
dependency startup without turning every prepend into a global wakeup.

View File

@ -0,0 +1,104 @@
# Roadmap and Research
Issue states change. Re-read each tracker before planning work and preserve
human ownership of issue and checklist state.
## Guiding Roadmap
- [piker #270](https://github.com/pikers/piker/issues/270): typed,
stream-composable FSP graph and DSL direction.
- [piker #107](https://github.com/pikers/piker/issues/107): cross-actor shared
memory tick/ring-buffer direction.
- [piker #216](https://github.com/pikers/piker/issues/216): exposing FSP output
through feed-like interfaces.
- [piker #325](https://github.com/pikers/piker/issues/325): faster streaming
y-range computation and overlap with FSP reductions.
- [piker #536](https://github.com/pikers/piker/issues/536): Arrow/Parquet data
ingest and columnar boundaries.
- [piker #98](https://github.com/pikers/piker/issues/98): older realtime feed
architecture discussion. Treat it as likely superseded; compare every idea
against current `Flume`, feed-bus, history, and sampler code.
- [Tractor #339](https://github.com/goodboy/tractor/issues/339): shared-memory
array and localhost IPC evolution.
- [hotbaud issues](https://github.com/guilledk/hotbaud/issues): experimental
high-throughput local transport and ring-buffer work.
## Near-Term Protocol Research
Prototype an explicit source revision envelope and measure it with two FSPs
and two overlaid feeds. The prototype should prove source-specific wakeups,
coalescing of consecutive prepends, stale-result rejection, and targeted UI
range updates. Keep the protocol independent of the final IPC transport.
Characterize historical repair strategies:
1. Full replay from the new first bound.
2. Prefix-only calculation for pointwise operators.
3. Replay from an operator checkpoint for stateful recurrences.
4. Bounded overlap repair for finite-window operators.
An FSP declaration eventually needs repair metadata describing locality,
state checkpointability, warm-up/overlap, and whether old inputs can alter all
later outputs.
## Candidate Libraries
Candidates are experiments, not dependency recommendations.
### Bounded Near-Term Prototypes
- [Bottleneck](https://github.com/pydata/bottleneck): compare rolling min/max,
rank, and moving-window kernels against NumPy/Numba for chart y-ranges and
finite-window FSPs.
- [River](https://github.com/online-ml/river): study stateful online estimator
APIs, drift detectors, and reference semantics. Do not assume its object
model belongs in the hottest tick path.
- [nanoarrow](https://github.com/apache/arrow-nanoarrow): evaluate lightweight
Arrow C data interfaces at native/process boundaries already motivated by
`pyarrow` and piker #536.
- [DuckDB](https://github.com/duckdb/duckdb): use for historical Parquet query,
replay slicing, and validation, outside realtime tick execution.
### Watchlist
- [hotbaud](https://github.com/guilledk/hotbaud): track API and correctness
maturity for localhost SHM/ring transport.
- [iceoryx2](https://github.com/eclipse-iceoryx/iceoryx2): study zero-copy
pub/sub ownership and lifecycle, but avoid competing with Tractor without a
narrow adapter benchmark.
- [DataFusion Python](https://github.com/apache/datafusion-python): Arrow-native
historical query and expression execution if existing Polars paths cannot
meet a demonstrated need.
- [Feldera](https://github.com/feldera/feldera): study DBSP/incremental-view
semantics for revision-aware graphs; its service/SQL runtime is not a direct
piker integration target.
- [Bytewax](https://github.com/bytewax/bytewax) and
[Arroyo](https://github.com/ArroyoSystems/arroyo): study dataflow semantics,
watermarking, and recovery, not as replacements for the actor tree.
GPU stacks such as JAX and CuPy are poor defaults for low-batch per-tick work
because transfer, compilation, and scheduling can dominate. Reconsider only
for measured large-batch research or many-market matrix workloads.
## Benchmark Design
Report separate distributions for:
- provider decode and normalization;
- source SHM write and bound publication;
- historical kernel and realtime recurrence;
- actor notification and stream fan-out;
- destination publication;
- visible-range formatting and Qt paint.
Use p50, p95, p99, maximum, allocation rate, and resident-memory growth. Test
steady state, burst load, backfill overlap, cancellation, and slow consumers.
Throughput without event-time correctness and bounded teardown is not a win.
## Medium-Term Sequence
First establish revisioned range messages and operator repair semantics. Then
decouple FSP orchestration from chart ownership and expose destination flows
through feed-like subscriptions. Next benchmark localhost transport adapters
behind the same protocol. Only after those contracts stabilize should graph
placement, fusion, worker scaling, and remote partitioning be automated.

View File

@ -0,0 +1,103 @@
# Financial Signal Processing
## Model the Clock First
Market inputs can be quote events, trades, book changes, fixed-time bars, or
revised historical bars. Name both event time and processing time. For
irregular events, either use time-aware recurrences or define the resampling
and interpolation policy explicitly.
Never let backfilled information leak into a value presented as historically
tradable. Distinguish corrected analysis history from the signal values that a
live strategy could have observed.
## Useful Technique Families
### Online Moments and Robust Statistics
Use Welford/Chan recurrences for stable online moments, compensated summation
for long accumulations, and mergeable summaries when partitioning actors.
Rolling median, quantiles, MAD, and winsorized measures are useful under heavy
tails, but exact rolling order statistics need more state than moments.
### Time-Aware Exponential Filters
For irregular event intervals, derive decay from elapsed time:
```text
alpha(dt) = 1 - exp(-dt / tau)
y_t = y_prev + alpha(dt) * (x_t - y_prev)
```
Record initialization, session reset, gap, and backfill behavior. A fixed
sample alpha is valid only after an explicit regularization policy.
### State-Space Models
Kalman and robust state-space filters fit latent price, spread, volatility,
and lead/lag estimates. Prefer square-root or Joseph-form covariance updates
when conditioning is poor. State revisions after backfill require replay from
a checkpoint or a bounded smoother; silently prepending observations without
replaying state is incorrect.
### Point Processes and Microstructure
Trade and quote arrivals are events, not merely sampled amplitudes. Consider
intensity, duration, signed order flow, imbalance, spread, queue change, and
self-excitation models. Validate against provider-specific aggregation and
duplicate/out-of-order behavior before interpreting a statistic.
### Volatility and Covariance
Realized variance, bipower variation, EW covariance, and range estimators can
be updated online. Asynchronous cross-market covariance needs synchronization
or estimators designed for non-synchronous observations; naive row alignment
creates lead/lag and Epps-effect artifacts.
### Spectral and Multiscale Methods
FFT methods assume a regular grid and a window. Declare detrending, tapering,
overlap, normalization, and latency. For tick data, resample deliberately or
use irregular-time methods. Wavelets and multiresolution filters can separate
horizons, but boundary handling and causal delay must be visible to strategy
code.
### Change and Anomaly Detection
CUSUM, Page-Hinkley, sequential likelihood ratios, and robust z-scores are
cheap online tools. Calibrate false alarms under dependence and regime shifts;
do not treat IID thresholds as market guarantees.
## Numerical Implementation Order
1. Establish a scalar reference with explicit state and semantics.
2. Vectorize historical bootstrap with NumPy.
3. Keep realtime updates as O(1) recurrences where possible.
4. Use Numba for measured numeric kernels with stable dtypes and no Python
object traffic.
5. Use Polars or Arrow for columnar historical transforms and interchange,
not automatically for each tick.
6. Partition by independent market/operator state before adding threads or a
new distributed runtime.
## Existing Stack
- NumPy 2.x: canonical dense and structured-array kernels.
- Numba: compiled CPU loops and recurrences that do not vectorize cleanly.
- Polars: parallel/lazy columnar history preparation and validation.
- PyArrow: columnar interchange, storage, and ingest boundaries.
- Tractor/Trio: structured distributed execution and lifecycle.
- PyQtGraph: visible-range rendering, not a compute scheduler.
## Algorithm Acceptance Criteria
Every new FSP should define:
- input fields, clock, ordering, duplicate, gap, and revision policy;
- output dtype, units, identity, and parameterization;
- warm-up length and initialization bias;
- causal latency and strategy-visible publication point;
- historical/realtime equivalence tolerance;
- reset and session-boundary behavior;
- complexity, state size, and representative throughput;
- replay behavior when historical source data changes.

View File

@ -0,0 +1,52 @@
# FSP Test Map
## Operator Tests
- Historical output schema, dtype, units, and source timestamp alignment.
- Realtime update equivalence with the final row of a batch recomputation.
- Warm-up, NaN, zero, gap, duplicate, and out-of-order behavior.
- Session reset and elapsed-time behavior.
- Parameter and source identity isolation.
## Engine Tests
- Equal absolute source/destination bounds.
- Normal one-step destination lag.
- Destination lead and lag greater than one step.
- Source prepend by one, two, and a full provider frame.
- In-place gap repair with unchanged source bounds.
- Equal lengths with shifted absolute bounds.
- Source change during historical recomputation and convergence afterward.
- Cancellation while the quote stream, sample stream, or compute generator is
blocked.
- Foreign FQME and timeframe backfill broadcasts.
## Graph and Actor Tests
- Dependency startup before downstream operator lookup.
- Actor-local flow registry isolation.
- One destination writer and multiple readers.
- Consumer cancellation without orphaned cascade or worker tasks.
- Worker failure propagation and SHM cleanup.
- Feed-like fan-out with a slow or disconnected consumer.
## UI Tests
- An `fsp_update` redraws every field visualization sharing its destination
token and no unrelated market, OHLC, or FSP visualization.
- A prepend outside the visible range does not move or re-range the live view.
- A revised value inside the visible range invalidates only required formatter
and graphics caches.
- Hidden charts defer rendering without stopping compute correctness.
## Performance Tests
- Historical bootstrap across realistic SHM capacities.
- O(1) realtime recurrence cost under bursty quotes.
- Consecutive provider-frame prepends with two or more cascades.
- Multi-market sampler broadcasts and targeted wakeups.
- Visible-range redraw cost at several units-per-pixel levels.
Prefer deterministic local arrays and controlled Trio task interleavings.
Use live providers only as a separate integration layer because credentials,
rate limits, sessions, and changing history make poor regression oracles.

View File

@ -0,0 +1 @@
../../.agents/skills/piker-clearing-expert

View File

@ -0,0 +1 @@
../../.agents/skills/piker-fsp-expert

1
.gitignore vendored
View File

@ -211,6 +211,7 @@ docs/conversations/
# BEGIN ai.skillz: runtime:open-wkt # BEGIN ai.skillz: runtime:open-wkt
.claude/wkts/ .claude/wkts/
claude_wkts claude_wkts
/wkts/
# END ai.skillz: runtime:open-wkt # END ai.skillz: runtime:open-wkt
# BEGIN ai.skillz: direct:symlink:claude:open-wkt # BEGIN ai.skillz: direct:symlink:claude:open-wkt

View File

@ -21,11 +21,14 @@ track new integration ideas and proposals in
## Shared Skills ## Shared Skills
Repo-specific skills use the portable Agent Skills Provider-neutral skill bundles should use the portable Agent Skills
frontmatter subset and live under `.claude/skills/` as a subset and live under `.agents/skills/` as their single source.
single source. Claude Code discovers that directory OpenCode discovers that project directory natively. Relative links in
natively, and OpenCode discovers the same project skills provider directories such as `.claude/skills/` may bridge clients which
without copies or generated wrappers. do not yet scan `.agents/skills/`; never copy the skill body.
Existing repo skills under `.claude/skills/` predate this convention
and can migrate separately without coupling their history to new skills.
Harness-specific command files should only delegate to a Harness-specific command files should only delegate to a
shared skill. They must not duplicate the skill body. shared skill. They must not duplicate the skill body.

View File

@ -8,6 +8,7 @@ integration for piker's shared coding-harness skills.
| Skill | Invocable | Description | | Skill | Invocable | Description |
|-------|-----------|-------------| |-------|-----------|-------------|
| [`commit-msg`](#commit-msg) | `/commit-msg` | Generate piker-style commit messages | | [`commit-msg`](#commit-msg) | `/commit-msg` | Generate piker-style commit messages |
| `piker-clearing-expert` | auto | Clearing, order-control, and accounting invariants |
| `piker-conc-expert` | auto | Actor-tree and structured-concurrency invariants | | `piker-conc-expert` | auto | Actor-tree and structured-concurrency invariants |
| `piker-profiling` | auto | `Profiler` API patterns for perf work | | `piker-profiling` | auto | `Profiler` API patterns for perf work |
| `piker-slang` | auto | Communication style + slang guide | | `piker-slang` | auto | Communication style + slang guide |
@ -19,10 +20,12 @@ Skills marked **auto** are background knowledge
applied automatically when Claude detects relevance. applied automatically when Claude detects relevance.
Only `commit-msg` is user-invoked via slash command. Only `commit-msg` is user-invoked via slash command.
Portable skill source files live under New portable skill source files live under
`.claude/skills/<skill-name>/SKILL.md` and are shared with `.agents/skills/<skill-name>/SKILL.md`. Relative links under
OpenCode. Claude-specific behavior belongs in command or `.claude/skills/` expose them to Claude Code without copying bodies.
settings files, not the shared skill bodies. Existing pre-migration skills can remain under `.claude/skills/`.
Claude-specific behavior belongs in command or settings files, not
shared skill bodies.
--- ---

View File

@ -1,14 +1,16 @@
# OpenCode Integration # OpenCode Integration
[OpenCode](https://opencode.ai/) uses piker's shared [OpenCode](https://opencode.ai/) discovers provider-neutral skills
repo-specific skills directly from `.claude/skills/`. directly from `.agents/skills/` and pre-migration piker skills from
There is no second copy of each skill to keep in sync. `.claude/skills/`. There is no second copy of each skill to keep in
sync.
## Available Skills ## Available Skills
| Skill | Activation | Specialization | | Skill | Activation | Specialization |
|-------|------------|----------------| |-------|------------|----------------|
| `commit-msg` | `/commit-msg` or explicit request | Piker commit-message style and artifacts | | `commit-msg` | `/commit-msg` or explicit request | Piker commit-message style and artifacts |
| `piker-clearing-expert` | automatic | Clearing, order-control, and accounting invariants |
| `piker-conc-expert` | automatic | `tractor` actor topology, RPC, and cancellation | | `piker-conc-expert` | automatic | `tractor` actor topology, RPC, and cancellation |
| `piker-profiling` | automatic | Cross-actor `Profiler` instrumentation | | `piker-profiling` | automatic | Cross-actor `Profiler` instrumentation |
| `piker-slang` | automatic | Project communication style | | `piker-slang` | automatic | Project communication style |
@ -49,3 +51,5 @@ hot-reloaded into an active session.
hardcode Claude Code attribution in shared workflows. hardcode Claude Code attribution in shared workflows.
- Keep supporting references beside each shared skill so - Keep supporting references beside each shared skill so
relative links resolve in every harness. relative links resolve in every harness.
- Put new provider-neutral bundles under `.agents/skills/`; use
relative provider links only when a harness requires one.

View File

@ -0,0 +1,36 @@
---
model: gpt-6 (exact variant not exposed)
service: codex
session: fsp-session-recovery-takeover-20260915
timestamp: 2026-09-15T15:52:29Z
git_ref: fadab3d2
scope: code
substantive: true
raw_file: 20260915T155229Z_fadab3d2_prompt_io.raw.md
---
## Prompt
Recover the interrupted OpenCode FSP session and take over its ongoing
work. The recovered child instruction records human authorization for
the minimal EMA omitted-default fix followed by a complete commit plan.
## Response summary
Removed the eager Numba signature that rejected Python's omitted
default arguments. Preserved nopython/nogil compilation and numerical
behavior. Added nine numerical regressions and verified all 38 focused
FSP cases. Rebuilt planning artifacts separately from production code.
## Files changed
- `piker/fsp/_momo.py`: drop the eager signature and unused type imports.
- `tests/test_fsp_momo.py`: omitted/explicit argument and seed coverage.
## Human edits
The human identified the omitted-default limitation in the earlier
session, authorized the minimal lazy-specialization approach, and
requested this takeover after OpenCode failed. Existing source changes
and older OpenCode provenance were retained. The new fix remains
unstaged and awaits human review; no acceptance is inferred.

View File

@ -0,0 +1,50 @@
---
model: gpt-6 (exact variant not exposed)
service: codex
timestamp: 2026-09-15T15:52:29Z
git_ref: fadab3d2
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
> i'm thinking maybe you can just take over that work and we can
> abandon that oc dialog for now?
Recovered child-session instructions describe the earlier human
authorization to remove EMA's eager Numba signature while retaining
`jit(nopython=True, nogil=True)`, verify defaults, and finish the commit
plan without staging or committing. This is recovered context, not a
verbatim quote of the earlier human authorization.
## Response
> `git diff HEAD~1..HEAD -- piker/fsp/_momo.py`
Removed EMA's eager Numba signature and now-unused Numba type imports.
Retained nopython/nogil compilation, the numerical kernel, historical
comments, and pre-existing staged and unstaged edits.
> `git diff HEAD~1..HEAD -- tests/test_fsp_momo.py`
Added production-dispatcher regressions for omitted positional and
keyword arguments, explicit None and numeric parameters, and the
previously corrected one-sample continuation. Expected outputs are
hand-computed recurrences, rather than the Python implementation.
Before the fix, the first case failed with:
```text
TypeError: No matching definition for argument type(s)
array(float64, 1d, C), omitted(default=None), omitted(default=None)
```
After the fix, the focused run outside the socket-restricted sandbox
reported:
```text
38 passed in 0.66s
```
Ruff passed for the modified momentum module and new regression file.
Combined-worktree results do not attest to isolated commit boundaries.

View File

@ -0,0 +1,10 @@
# AI Prompt I/O Log — codex
This directory tracks prompts and outputs for AI-assisted development
using Codex. Substantive contributions include model identification,
prompt context, output records, and human contribution accounting.
Code output uses Git diff references; non-code output is retained in
the paired `.raw.md` file. Human contributors remain responsible for
review and acceptance. Entries follow the
[NLNet generative AI policy](https://nlnet.nl/foundation/policies/generativeAI/).

View File

@ -0,0 +1,35 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: opencode-pid-469264
timestamp: 2026-08-28T01:30:57Z
git_ref: 6ca6ab2a
scope: code
substantive: true
raw_file: 20260828T013057Z_6ca6ab2a_prompt_io.raw.md
---
## Prompt
Investigate Piker's clearing, accounting, order-control, and
trade-ledger flows in depth while fixing a paper-engine crash where
RapidFuzz called `len()` on a `MktPair` during ledger persistence.
## Response summary
Changed `SymbologyCache.search()` to fuzzy-match table keys instead of
struct values, preserving matched aliases and preventing the paper
trade dialog from crashing when qualifying legacy ledger FQMEs. Added
an end-to-end regression for the implicated ledger-write path.
## Files changed
- `piker/data/_symcache.py` - searches string keys and resolves values.
- `tests/test_accounting.py` - covers aliases and paper-ledger writes.
## Human edits
The human supplied the production traceback, identified the affected
subsystems, and directed an implementation-level fix with deep system
familiarization. No follow-up correction or direct manual source edit
was identified during this interaction.

View File

@ -0,0 +1,44 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-28T01:30:57Z
git_ref: 6ca6ab2a
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
Become familiar with Piker's clearing, accounting, order-control,
and trade-ledger tracking subsystems in deep detail while fixing a
paper-engine crash. The supplied traceback showed
`TransactionLedger.write_config()` calling
`SymbologyCache.search()`, followed by RapidFuzz raising
`TypeError: object of type 'MktPair' has no len()` after a paper fill.
## Response
Fixed the crash at the symbology-cache contract boundary.
`SymbologyCache.search()` now fuzzy-matches the selected table's
string keys and resolves each matched key back to its struct value.
RapidFuzz therefore never receives `MktPair` values as sequences,
and native market-ID aliases remain intact in search results.
> `git diff HEAD~1..HEAD -- piker/data/_symcache.py`
Added a regression that recreates the failing paper-ledger write with
an unqualified Binance perpetual FQME. It also checks a native market-ID
alias, preventing a weaker `processor=str` workaround that would lose
the actual matched key.
> `git diff HEAD~1..HEAD -- tests/test_accounting.py`
Verification:
`./py313/bin/python -m pytest -p no:xonsh -q -x --tb=short
--no-header tests/test_accounting.py::test_account_file_default_empty
tests/test_accounting.py::test_paper_ledger_fuzzy_qualifies_mktmap_keys`
Result: `2 passed in 0.02s`.
`git diff --check` also passed. Ruff was not available in the existing
`py313` environment, so no environment provisioning was performed.

View File

@ -0,0 +1,46 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: opencode-pid-469264
timestamp: 2026-08-28T02:14:11Z
git_ref: 6ca6ab2a
scope: config
substantive: true
raw_file: 20260828T021411Z_6ca6ab2a_prompt_io.raw.md
---
## Prompt
Move the fix out of the root checkout into an isolated worktree and
include a Piker-specific clearing, accounting, order-control, and
trade-ledger expert skill in the worktree's commit plan.
Use a cross-harness, AI-provider-neutral standard location for its
canonical files instead of making one provider directory authoritative.
## Response summary
Created a portable `piker-clearing-expert` skill bundle under the
provider-neutral `.agents/skills/` convention, with a relative Claude
compatibility link. It covers subsystem ownership, protocols,
identities, persistence, backend variation, debugging hazards, and
test strategy. Added the canonical `wkts/` runtime ignore.
## Files changed
- `.gitignore` - ignores the canonical root-level worktree directory.
- `.agents/skills/piker-clearing-expert/SKILL.md` - core expertise.
- `.agents/skills/piker-clearing-expert/architecture.md` - deep flows.
- `.agents/skills/piker-clearing-expert/gotchas.md` - diagnosis.
- `.agents/skills/piker-clearing-expert/test-map.md` - test map.
- `.claude/skills/piker-clearing-expert` - relative Claude bridge.
- `ai/README.md` - defines the provider-neutral source convention.
- `ai/claude-code/README.md` - advertises the shared skill to Claude.
- `ai/opencode/README.md` - advertises the shared skill to OpenCode.
## Human edits
The human rejected planning in the dirty root checkout, required a
dedicated worktree, expanded the work to include a durable expert
skill, and then directed its canonical source into a cross-provider
standard location. The agent applied those directed scope and layout
changes; no direct manual source edit was identified.

View File

@ -0,0 +1,62 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-28T02:14:11Z
git_ref: 6ca6ab2a
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
Move the symcache/accounting fix into an isolated worktree and, as
part of its commit plan, distill a Piker-specific skill for deep
clearing, accounting, order-control, and trade-ledger expertise.
A follow-up required the skill's canonical files to begin moving into
a cross-harness, AI-provider-neutral standard location.
## Response
Created the shared repository skill bundle under the provider-neutral
`.agents/skills/piker-clearing-expert/` location. Its main skill defines
activation cues,
runtime ownership, event and identifier contracts, accounting and
persistence invariants, market identity rules, an editing workflow,
and a canonical source map.
> `git diff HEAD~1..HEAD -- .agents/skills/piker-clearing-expert/SKILL.md`
Added a detailed architecture reference for EMS routing, ack/cancel
ordering, paper and live execution, ledger normalization, position
events, symbology, and context-managed persistence.
> `git diff HEAD~1..HEAD -- .agents/skills/piker-clearing-expert/architecture.md`
Added symptom-driven gotchas and debugging breadcrumbs for actor
teardown, ID skew, position reconciliation, duplicate fills,
paper/live differences, and RapidFuzz mapping misuse.
> `git diff HEAD~1..HEAD -- .agents/skills/piker-clearing-expert/gotchas.md`
Added a deterministic-to-live test map and regression-design guidance.
> `git diff HEAD~1..HEAD -- .agents/skills/piker-clearing-expert/test-map.md`
Updated the worktree runtime ignore block to include the canonical
root-level `wkts/` directory required by the worktree lifecycle.
> `git diff HEAD~1..HEAD -- .gitignore`
Added a relative Claude compatibility link and documented the
provider-neutral source convention in the shared and harness-specific
inventories.
> `git diff HEAD~1..HEAD -- .claude/skills/piker-clearing-expert`
> `git diff HEAD~1..HEAD -- ai/README.md`
> `git diff HEAD~1..HEAD -- ai/claude-code/README.md`
> `git diff HEAD~1..HEAD -- ai/opencode/README.md`
The skill cross-links `piker-conc-expert` for actor-runtime mechanics
instead of duplicating structured-concurrency guidance.

View File

@ -0,0 +1,35 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: pid-469264
timestamp: 2026-08-28T06:31:26Z
git_ref: 1606122b
scope: code
substantive: true
raw_file: 20260828T063126Z_1606122b_prompt_io.raw.md
---
## Prompt
Review the staged IB broker patch before committing it. After the review
identified a deterministic error-parser crash, continue with the available
next steps while leaving the typed EMS worktree and Ruff change untouched.
## Response summary
Repaired bracketed IB API error parsing by splitting on the matched error
prefix. Added a deterministic regression that routes bracketed and plain
strings through the real trade-event relay and verifies their EMS error
messages.
## Files changed
- `piker/brokers/ib/broker.py` - safely extracts bracketed API error codes.
- `tests/test_ib_broker.py` - covers bracketed and plain error relay inputs.
## Human edits
The human required review before the root commit, accepted the identified
parser defect as the next work item, and explicitly authorized continuation.
The pre-existing staged IB refactor is human-authored; the agent applied only
the parser repair and focused regression described above.

View File

@ -0,0 +1,30 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-28T06:31:26Z
git_ref: fix_symcache_search
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
The human asked for a review of the root checkout's staged IB patch before
committing it. The review found that bracketed API error strings call
`str.partition()` without its required separator. The human then instructed
the agent to continue with available next steps, or ask only if unsure.
## Response
> `git diff HEAD~1..HEAD -- piker/brokers/ib/broker.py`
The generated parser repair uses the recognized `[code ` or `[Errno ` prefix
to extract the numeric code without changing the surrounding event relay.
> `git diff HEAD~1..HEAD -- tests/test_ib_broker.py`
The generated regression drives bracketed and plain strings through
`deliver_trade_events()` and verifies that one `BrokerdError` reaches EMS
with the expected reason and unknown request ID.
The typed EMS worktree and the unrelated Ruff policy change were left
untouched. No commit was created.

View File

@ -0,0 +1,67 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: opencode-fsp-backfill-sync-20260828
timestamp: 2026-08-28T19:50:32+00:00
git_ref: 9115d24d
scope: code
substantive: true
raw_file: 20260828T195032Z_9115d24d_prompt_io.raw.md
---
## Prompt
The human requested a repo-specialist role for realtime financial signal
processing, a new FSP expert skill, and a fix for repeated FSP desynchronizing
and whole-chart refreshes during near-term backfill with multiple overlaid
market feeds. Work was required in a new worktree based on the root checkout.
Follow-up direction required explicit typing throughout the patch, exact
Qt/pyqtgraph widget types, project-style union spacing, and a final
per-behaviour multi-commit plan.
## Response summary
The response traced the reported `step_diff=0`, `len_diff=2000` state to
legitimate source-history prepends. It retained required historical
recomputation while preventing backfill wakeups from advancing destination
rows or refreshing unrelated charts. It introduced immutable bootstrap
snapshots, source/timeframe event filtering, absolute SHM-bound checks,
targeted graphics invalidation, and deterministic tests.
The typing follow-up established a canonical FSP generator protocol, removed
an engine-side cast, normalized the `wma` and `rsi` built-ins, validated
historical output schemas, and made generator cleanup deterministic. A later
review replaced arbitrary redraw kwargs with an explicit control, typed the
`Viz`/`PlotItem`/`Axis`/`YAxisLabel` path, added normalized quote and tick
shapes, and corrected false iterator and NumPy/Numba annotations.
The new provider-neutral `piker-fsp-expert` skill captures architecture,
mathematical methods, optimization guidance, roadmap context, dependency
research, and a test matrix.
## Files changed
- `piker/fsp/_api.py` - typed FSP decorator and generator contract.
- `piker/fsp/_engine.py` - backfill synchronization, snapshot, publication,
event filtering, and cleanup behavior.
- `piker/fsp/_momo.py` - protocol-conforming WMA and RSI streams.
- `piker/fsp/_volume.py` - typed volume FSP stream declarations.
- `piker/data/ticktools.py` - concrete quote/tick and iterator contracts.
- `piker/data/_sharedmem.py` - serializable SHM token message shape.
- `piker/ui/_dataviz.py` - piker `PlotItem` ownership on `Viz`.
- `piker/ui/_pg_overrides.py` - piker view and axis return types.
- `piker/ui/_fsp.py` - destination-token-targeted redraw and y-range refresh.
- `tests/test_fsp_sync.py` - deterministic engine, UI, and operator tests.
- `.agents/skills/piker-fsp-expert/` - reusable FSP specialist knowledge.
- `.claude/skills/piker-fsp-expert` - Claude-compatible skill link.
## Human edits
The human identified insufficient typing in the initial patch and required
the implementation to expose and repair the actual FSP protocol instead of
using broad or misleading casts. In later review, the human specifically
rejected `**kwargs: Any`, an optional `Viz.shm`, a broad sticky-label type,
and spaced union syntax, then directed the agent to infer reusable types from
the actual graphics widgets. The human also required a per-behaviour commit
plan for final review. No direct manual source edits were observed.

View File

@ -0,0 +1,54 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-28T19:50:32+00:00
git_ref: 9115d24d
diff_cmd: git diff HEAD~1..HEAD
---
The response implemented and analyzed the requested FSP backfill fix.
> `git diff HEAD~1..HEAD -- piker/fsp/_api.py`
>
> `git diff HEAD~1..HEAD -- piker/fsp/_engine.py`
>
> `git diff HEAD~1..HEAD -- piker/fsp/_momo.py`
>
> `git diff HEAD~1..HEAD -- piker/fsp/_volume.py`
>
> `git diff HEAD~1..HEAD -- piker/data/ticktools.py`
>
> `git diff HEAD~1..HEAD -- piker/data/_sharedmem.py`
>
> `git diff HEAD~1..HEAD -- piker/ui/_dataviz.py`
>
> `git diff HEAD~1..HEAD -- piker/ui/_pg_overrides.py`
>
> `git diff HEAD~1..HEAD -- piker/ui/_fsp.py`
>
> `git diff HEAD~1..HEAD -- tests/test_fsp_sync.py`
>
> `git diff HEAD~1..HEAD -- .agents/skills/piker-fsp-expert/`
The generated implementation distinguishes history revisions from sample
advancement, filters foreign backfill events, computes historical output from
an immutable source snapshot, and targets FSP-backed visualizations for cache
invalidation, redraw, and y-range updates. It also establishes a typed FSP
stream protocol, normalizes the dormant scalar built-ins to that protocol,
and adds deterministic regression coverage. The typing review replaces
arbitrary redraw kwargs with an explicit boolean control, records the
concrete piker graphics object chain, and introduces reusable normalized
quote, tick, and SHM-token message shapes.
The generated skill bundle records the FSP runtime architecture, financial
signal-processing principles, testing matrix, roadmap issues, and candidate
numerics and IPC research projects.
Verification reported by the response:
- focused FSP and deterministic history tests passed;
- Ruff passed for all changed Python files;
- focused `ty` checks passed for tick and scalar FSP contracts;
- compileall passed for all changed Python files;
- `git diff --check` passed;
- repeated read-only code review found no remaining actionable findings.

View File

@ -0,0 +1,41 @@
---
model: gpt-5.6-sol
service: opencode
session: fsp-backfill-sync-live-volume
timestamp: 2026-09-03T23:25:25Z
git_ref: fadab3d2
scope: code
substantive: true
raw_file: 20260903T232525Z_fadab3d2_prompt_io.raw.md
---
## Prompt
Diagnose and fix MNQ's same-index volume repaint, missing post-closure
volume curve, stale zoomed-out downsample path, and repeated
`No yrange provided for dolla_vlm` warnings using live SHM evidence.
## Response summary
Declared source sample periods on `Flume` so FSP cascades cannot mistake
a market-closure timestamp gap for their regular cadence. Clamped
negative provider volume at sampling and volume-FSP boundaries, honored
M4 renderer cache invalidation, and skipped unchanged realtime FSP
redraws while retaining forced history repairs.
## Files changed
- `piker/data/flows.py` - declare RT/history sample periods.
- `piker/data/_sampling.py` - prevent negative sampled volume.
- `piker/fsp/_engine.py` - subscribe using the declared RT period.
- `piker/fsp/_volume.py` - clamp volume history and trade sizes.
- `piker/ui/_dataviz.py` - honor renderer cache resets.
- `piker/ui/_fsp.py` - skip identical realtime FSP redraws.
- `tests/test_fsp_sync.py` - cover the observed regressions.
## Human edits
The human identified each visual failure, supplied the exact console
warning, confirmed that the later-started `flow_rates` cascade behaved
correctly, requested declared cadence instead of timestamp inference,
performed chart restarts, and live-validated the final behavior.

View File

@ -0,0 +1,71 @@
---
model: gpt-5.6-sol
service: opencode
timestamp: 2026-09-03T23:25:25Z
git_ref: fadab3d2
diff_cmd: git diff HEAD~4..HEAD
---
## Prompt
The human reported that the MNQ volume subchart repeatedly rendered
its last bar at one index around the equity-futures closure, then
reported that the volume curve disappeared after reopening. During
live testing, the human confirmed that `flow_rates`, which started
later, advanced correctly while `dolla_vlm` remained pinned. The human
then reported a stale downsampled curve on zoom-out and supplied the
repeated console warning:
`No yrange provided for dolla_vlm!?`
The human requested a live diagnosis and fix, validated each restart,
and confirmed the final chart behavior was fixed.
## Response
Read-only SHM snapshots showed that `dolla_vlm` remained at exclusive
bound `172801` while the source advanced hundreds of rows. Its one
current row kept receiving current quote values, explaining the
same-index repaint. `flow_rates` remained aligned because it started
after regular one-second timestamps resumed.
The cascade had inferred its sampler period from the last two distinct
timestamps. A market closure therefore looked like the source cadence.
Explicit `Flume` RT/history periods now drive sampling and downsampling
metadata instead.
> `git diff HEAD~4..HEAD -- piker/data/flows.py`
> `git diff HEAD~4..HEAD -- piker/fsp/_engine.py`
Further SHM inspection found an IB source row with volume `-2319096`.
`dolla_vlm` multiplied that value into a `-68277471626` outlier, causing
`Viz.maxmin()` to reject the zoomed-out range. Sample ingestion and
volume FSP history/realtime calculations now clamp negative volume.
> `git diff HEAD~4..HEAD -- piker/data/_sampling.py`
> `git diff HEAD~4..HEAD -- piker/fsp/_volume.py`
The M4 renderer already returned a cache-reset signal after rebuilding
a downsampled path, but `Viz.update_graphics()` overwrote that signal
with `False`. The override was removed so Qt's device-coordinate cache
is invalidated with the new path.
> `git diff HEAD~4..HEAD -- piker/ui/_dataviz.py`
Normal quote cycles also redrew unchanged FSP state. A per-`Viz`
bounds/timestamp/value signature now suppresses only identical
realtime refreshes; forced history redraws still bypass it.
> `git diff HEAD~4..HEAD -- piker/ui/_fsp.py`
Regression coverage models closure-gap cadence, negative provider
volume, unchanged realtime redraws, and the broader FSP synchronization
behavior.
> `git diff HEAD~4..HEAD -- tests/test_fsp_sync.py`
Live verification after restart showed source and `dolla_vlm` advancing
at the same bound, zero negative destination rows, and correct zoomed-out
rendering. The focused suite completed with 38 passing tests.

View File

@ -0,0 +1,40 @@
---
model: gpt-5.6-sol
service: opencode
session: tuicr-fsp-api-review
timestamp: 2026-09-08T02:06:19Z
git_ref: fadab3d2
scope: code
substantive: true
raw_file: 20260908T020619Z_fadab3d2_prompt_io.raw.md
---
## Prompt
Address all 14 comments from the local staged FSP Tuicr review, account
for relevant unstaged behavior changes, verify the fixes, and add local
responses without staging or committing repository changes.
## Response summary
Clarified typed wire-schema and generic-decorator intent, adopted more
specific FSP and feed type names, made momentum array typing explicit,
documented the WAP tuple fields, and verified both FSP regressions and
the production Numba call shape.
## Files changed
- `piker/data/__init__.py` - export the renamed feed quote type.
- `piker/data/ticktools.py` - clarify wire dicts and future structs.
- `piker/fsp/_api.py` - clarify history and decorator types.
- `piker/fsp/_engine.py` - consume the renamed public types.
- `piker/fsp/_momo.py` - make array and stream types explicit.
- `piker/fsp/_volume.py` - document WAP return fields.
- `tests/test_fsp_sync.py` - consume the renamed feed quote type.
## Human edits
The human reviewed the staged API boundary in Tuicr, requested the type
renames and documentation, rejected an opaque NumPy alias, identified
missing local annotations, required per-field tuple comments, and asked
that existing unstaged fixes be considered when responding.

View File

@ -0,0 +1,47 @@
---
model: gpt-5.6-sol
service: opencode
timestamp: 2026-09-08T02:06:19Z
git_ref: fadab3d2
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
The human supplied 14 comments from the local Tuicr review session
`fsp_backfill_sync@wkt-fsp_backfill_sync/staged/fadab3d` and invoked
`/code-review-changes -r`. They asked that comments already addressed
by unstaged behavioral work be recognized when choosing each response.
## Response
The review was correlated one-to-one with its persisted Tuicr records.
The resulting adjustments clarify the existing dict wire schema and a
future paired `msgspec.Struct` migration, and rename the broad `Quote`
and `HistoryOutput` types to `FeedQuote` and `FspHistory`.
> `git diff HEAD~1..HEAD -- piker/data/__init__.py`
> `git diff HEAD~1..HEAD -- piker/data/ticktools.py`
> `git diff HEAD~1..HEAD -- piker/fsp/_api.py`
> `git diff HEAD~1..HEAD -- piker/fsp/_engine.py`
The FSP decorator now documents its PEP 695 parameter specification and
its bare and configured return forms. Momentum types use explicit
`NDArray[np.float64]` annotations, and WMA locals and the ignored tick
target are named and typed explicitly. The WAP tuple return is split
across documented fields.
> `git diff HEAD~1..HEAD -- piker/fsp/_momo.py`
> `git diff HEAD~1..HEAD -- piker/fsp/_volume.py`
> `git diff HEAD~1..HEAD -- tests/test_fsp_sync.py`
The focused FSP regression file passed with 29 tests. Ruff, Ruff's E501
line-length selection, compileall, and diff whitespace checks passed.
The production Numba call shape was also executed successfully with all
three arguments; its pre-existing explicit signature does not accept
omitted defaults.

View File

@ -52,7 +52,6 @@ from ib_async.objects import (
from piker import config from piker import config
from piker.log import ( from piker.log import (
get_logger, get_logger,
get_console_log,
) )
from piker.types import Struct from piker.types import Struct
from piker.accounting import ( from piker.accounting import (
@ -1429,27 +1428,35 @@ async def deliver_trade_events(
# NOTE: see impl deats in # NOTE: see impl deats in
# `Client.inline_errors()::push_err()` # `Client.inline_errors()::push_err()`
err: dict|str = item err: dict|str = item
is_msg: bool = False
# std case, never relay errors for non-order-control # std case, never relay errors for non-order-control
# related issues. # related issues.
# https://interactivebrokers.github.io/tws-api/message_codes.html # https://interactivebrokers.github.io/tws-api/message_codes.html
if isinstance(err, dict): match err:
case dict():
# if isinstance(err, dict):
code: int = err['error_code'] code: int = err['error_code']
reason: str = err['reason'] reason: str = err['reason']
reqid: str = str(err['reqid']) reqid: str = str(err['reqid'])
is_msg = True
case str():
# XXX, sometimes you'll get just a `str` of the form, # XXX, sometimes you'll get just a `str` of the form,
# '[code 104] connection failed' or something.. # '[code 104] connection failed' or something..
elif isinstance(err, str): # elif isinstance(err, str):
code: int|str = '<NA-err-code>'
code_part, _, reason = err.rpartition(']') code_part, _, reason = err.rpartition(']')
if code_part: if code_part:
for prefix_patt in [ for prefix_patt in [
'[Errno ', '[Errno ',
'[code ', '[code ',
]: ]:
code_part, _, code = code_part.partition() _, prefix, code_str = (
if code: code_part.partition(prefix_patt)
code = int(code) )
if prefix:
code = int(code_str)
break break
reqid: str = '<unknown>' reqid: str = '<unknown>'
@ -1497,16 +1504,17 @@ async def deliver_trade_events(
) )
continue continue
if err['reqid'] == -1: if (
is_msg
and
err['reqid'] == -1
):
log.error( log.error(
f'TWS external order error ??\n' f'TWS external order error ??\n'
f'{pformat(err)}\n' f'{pformat(err)}\n'
) )
flow: dict = dict( flow: dict = dict(flows.get(reqid) or {})
flows.get(reqid)
or {}
)
# TODO: we don't want to relay data feed / lookup errors # TODO: we don't want to relay data feed / lookup errors
# so we need some further filtering logic here.. # so we need some further filtering logic here..

View File

@ -22,7 +22,11 @@ and storing data from your brokers as well as
sharing live streams over a network. sharing live streams over a network.
""" """
from .ticktools import iterticks from .ticktools import (
FeedQuote,
iterticks,
Tick,
)
from tractor.ipc._shm import ( from tractor.ipc._shm import (
ShmArray, ShmArray,
get_shm_token, get_shm_token,
@ -54,7 +58,9 @@ __all__: list[str] = [
'Feed', 'Feed',
'open_feed', 'open_feed',
'ShmArray', 'ShmArray',
'FeedQuote',
'iterticks', 'iterticks',
'Tick',
'maybe_open_shm_array', 'maybe_open_shm_array',
'match_from_pairs', 'match_from_pairs',
'attach_shm_array', 'attach_shm_array',

View File

@ -670,7 +670,10 @@ async def sample_and_broadcast(
['open', 'high', 'low', 'volume'] ['open', 'high', 'low', 'volume']
] ]
new_v: float = tick.get('size', 0) new_v: float = max(
float(tick.get('size', 0)),
0.,
)
if v == 0 and new_v: if v == 0 and new_v:
# no trades for this bar yet so the open # no trades for this bar yet so the open
@ -678,11 +681,14 @@ async def sample_and_broadcast(
o = last o = last
if sum_tick_vlm: if sum_tick_vlm:
volume = v + new_v volume = max(float(v), 0.) + new_v
else: else:
# presume backend takes care of summing # presume backend takes care of summing
# it's own vlm # it's own vlm
volume = quote['volume'] volume = max(
float(quote['volume']),
0.,
)
shm.array[[ shm.array[[
'open', 'open',

View File

@ -26,6 +26,8 @@ are now imported directly from tractor throughout
the codebase. the codebase.
''' '''
from typing import TypedDict
import numpy as np import numpy as np
from tractor.ipc._shm import ( from tractor.ipc._shm import (
@ -39,6 +41,20 @@ from tractor.ipc._shm import (
from ._util import log from ._util import log
type DTypeField = (
tuple[str, str]
|tuple[str, str, tuple[int, ...]]
)
class NDTokenMsg(TypedDict):
shm_name: str
shm_first_index_name: str
shm_last_index_name: str
dtype_descr: tuple[DTypeField, ...]
size: int
key: str|None
def _make_token( def _make_token(
key: str, key: str,
@ -162,7 +178,7 @@ def maybe_open_shm_array(
def try_read( def try_read(
array: np.ndarray, array: np.ndarray,
) -> np.ndarray|None: ) -> np.void|None:
''' '''
Try to read the last row from a shared mem Try to read the last row from a shared mem
array or ``None`` if the array read returns array or ``None`` if the array read returns

View File

@ -334,15 +334,16 @@ class SymbologyCache(Struct):
matches in a `dict` including the `MktPair` values. matches in a `dict` including the `MktPair` values.
''' '''
entries: dict[str, Struct] = getattr(self, table)
matches = fuzzy.extract( matches = fuzzy.extract(
pattern, pattern,
getattr(self, table), list(entries),
score_cutoff=50, score_cutoff=50,
) )
# repack in dict[fqme, MktPair] form # repack in dict[fqme, MktPair] form
return { return {
item[0].fqme: item[0] item[0]: entries[item[0]]
for item in matches for item in matches
} }

View File

@ -24,7 +24,6 @@ real-time data processing data-structures.
from __future__ import annotations from __future__ import annotations
import tractor import tractor
import pendulum
import numpy as np import numpy as np
from piker.types import Struct from piker.types import Struct
@ -73,6 +72,8 @@ class Flume(Struct):
izero_hist: int = 0 izero_hist: int = 0
izero_rt: int = 0 izero_rt: int = 0
throttle_rate: int | None = None throttle_rate: int | None = None
rt_sample_period_s: float = 1.
hist_sample_period_s: float = 60.
@property @property
def rt_shm(self) -> ShmArray: def rt_shm(self) -> ShmArray:
@ -115,17 +116,9 @@ class Flume(Struct):
period and ratio between them. period and ratio between them.
''' '''
times: np.ndarray = self.hist_shm.array['time'] rt_step_size_s: float = self.rt_sample_period_s
end: float | int = pendulum.from_timestamp(times[-1]) hist_step_size_s: float = self.hist_sample_period_s
start: float | int = pendulum.from_timestamp(times[times != times[-1]][-1]) ratio: float = hist_step_size_s / rt_step_size_s
hist_step_size_s: float = (end - start).seconds
times = self.rt_shm.array['time']
end = pendulum.from_timestamp(times[-1])
start = pendulum.from_timestamp(times[times != times[-1]][-1])
rt_step_size_s = (end - start).seconds
ratio = hist_step_size_s / rt_step_size_s
return ( return (
rt_step_size_s, rt_step_size_s,
hist_step_size_s, hist_step_size_s,

View File

@ -20,10 +20,31 @@ Tick event stream processing, filter-by-types, format-normalization.
''' '''
from itertools import chain from itertools import chain
from typing import ( from typing import (
Any, Iterable,
AsyncIterator, Iterator,
Sequence,
TypedDict,
) )
# TODO: migrate feed emitters and consumers to ``msgspec.Struct``
# together. These ``TypedDict`` types only describe the builtin dicts
# currently transported over IPC; changing only the receiver-side
# annotations would not change serialization.
class Tick(TypedDict, total=False):
type: str
price: float
size: float
time: float
class FeedQuote(TypedDict, total=False):
ticks: list[Tick]
tradeRate: float
volumeRate: float
broker_ts: float
brokerd_ts: float
# tick-type-classes template for all possible "lowest level" events # tick-type-classes template for all possible "lowest level" events
# that can can be emitted by the "top of book" L1 queues and # that can can be emitted by the "top of book" L1 queues and
# price-matching (with eventual clearing) in a double auction # price-matching (with eventual clearing) in a double auction
@ -41,15 +62,12 @@ _auction_ticks: set[str] = set.union(*_tick_groups.values())
def frame_ticks( def frame_ticks(
quote: dict[str, Any], quote: FeedQuote,
ticks_by_type: dict | None = None, ticks_by_type: dict[str, list[Tick]]|None = None,
ticks_in_order: list[dict[str, Any]] | None = None ticks_in_order: list[Tick]|None = None,
) -> dict[ ) -> dict[str, list[Tick]]:
str,
list[dict[str, Any]]
]:
''' '''
XXX: build a tick-by-type table of lists XXX: build a tick-by-type table of lists
of tick messages. This allows for less of tick messages. This allows for less
@ -82,7 +100,7 @@ def frame_ticks(
# append in reverse FIFO order for in-order iteration on # append in reverse FIFO order for in-order iteration on
# receiver side. # receiver side.
tick: dict[str, Any] tick: Tick
for tick in ticks: for tick in ticks:
tbt.setdefault( tbt.setdefault(
tick['type'], tick['type'],
@ -104,8 +122,8 @@ def frame_ticks(
def iterticks( def iterticks(
quote: dict, quote: FeedQuote,
types: tuple[str] = ( types: Sequence[str] = (
'trade', 'trade',
'dark_trade', 'dark_trade',
), ),
@ -116,7 +134,7 @@ def iterticks(
# with this? # with this?
frame_by_type: bool = False, frame_by_type: bool = False,
) -> AsyncIterator: ) -> Iterator[Tick]:
''' '''
Iterate through ticks delivered per quote cycle, filter and Iterate through ticks delivered per quote cycle, filter and
yield any declared in `types`. yield any declared in `types`.
@ -163,10 +181,13 @@ def iterticks(
ticks.extend(list(chain(trades.values(), darks.values()))) ticks.extend(list(chain(trades.values(), darks.values())))
# most-recent-first # most-recent-first
if reverse: ticks_iter: Iterable[Tick] = (
ticks = reversed(ticks) reversed(ticks)
if reverse
else ticks
)
for tick in ticks: for tick in ticks_iter:
# print(f"{quote['symbol']}: {tick}") # print(f"{quote['symbol']}: {tick}")
ttype = tick.get('type') ttype = tick.get('type')
if ttype in types: if ttype in types:

View File

@ -25,12 +25,11 @@ FSP (financial signal processing) apis.
# - composition of fsps / implicit chaining syntax (we need an issue) # - composition of fsps / implicit chaining syntax (we need an issue)
from __future__ import annotations from __future__ import annotations
from functools import partial
from typing import ( from typing import (
Any, AsyncGenerator,
Callable, Callable,
Awaitable, overload,
Optional, Protocol,
) )
import numpy as np import numpy as np
@ -47,11 +46,32 @@ from ..log import get_logger
log = get_logger(__name__) log = get_logger(__name__)
type FspHistory = dict[str, np.ndarray|None]|np.ndarray
type RealtimeValue = np.ndarray|np.number|int|float
type FspYield = FspHistory|tuple[str, RealtimeValue]
type FspStream = AsyncGenerator[FspYield, None]
type FspConfigValue = str|bool|int|float
# ``**P`` declares a PEP 695 ``ParamSpec`` so wrappers retain each
# decorated FSP's positional and keyword parameter types:
# https://docs.python.org/3/reference/compound_stmts.html#type-params
class FspFunc[**P](Protocol):
@property
def __name__(self) -> str: ...
def __call__(
self,
*args: P.args,
**kwargs: P.kwargs,
) -> FspStream: ...
# global fsp registry filled out by @fsp decorator below # global fsp registry filled out by @fsp decorator below
_fsp_registry = {} _fsp_registry: dict[NamespacePath, Fsp] = {}
def _load_builtins() -> dict[tuple, Callable]: def _load_builtins() -> dict[NamespacePath, Fsp]:
# import to implicity trigger registration via ``@fsp`` # import to implicity trigger registration via ``@fsp``
from . import _momo # noqa from . import _momo # noqa
@ -60,7 +80,7 @@ def _load_builtins() -> dict[tuple, Callable]:
return _fsp_registry return _fsp_registry
class Fsp: class Fsp[**P]:
''' '''
"Financial signal processor" decorator wrapped async function. "Financial signal processor" decorator wrapped async function.
@ -79,28 +99,28 @@ class Fsp:
# shm flow. # shm flow.
_flow_registry: dict[ _flow_registry: dict[
tuple[NDToken, str], tuple[NDToken, str],
tuple[NDToken, Optional[ShmArray]], tuple[NDToken, ShmArray|None],
] = {} ] = {}
def __init__( def __init__(
self, self,
func: Callable[..., Awaitable], func: FspFunc[P],
*, *,
outputs: tuple[str] = (), outputs: tuple[str, ...] = (),
display_name: Optional[str] = None, display_name: str|None = None,
**config, **config: FspConfigValue,
) -> None: ) -> None:
# TODO (maybe): # TODO (maybe):
# - type introspection? # - type introspection?
# - should we make this a wrapt object proxy? # - should we make this a wrapt object proxy?
self.func = func self.func: FspFunc[P] = func
self.__name__ = func.__name__ # XXX: must have func-object name self.__name__ = func.__name__ # XXX: must have func-object name
self.ns_path: tuple[str, str] = NamespacePath.from_ref(func) self.ns_path: NamespacePath = NamespacePath.from_ref(func)
self.outputs = outputs self.outputs = outputs
self.config: dict[str, Any] = config self.config: dict[str, FspConfigValue] = config
# register with declared set. # register with declared set.
_fsp_registry[self.ns_path] = self _fsp_registry[self.ns_path] = self
@ -116,9 +136,10 @@ class Fsp:
# type annots from pep 612: # type annots from pep 612:
# https://www.python.org/dev/peps/pep-0612/ # https://www.python.org/dev/peps/pep-0612/
# instance, # instance,
*args, *args: P.args,
**kwargs **kwargs: P.kwargs,
):
) -> FspStream:
return self.func(*args, **kwargs) return self.func(*args, **kwargs)
def get_shm( def get_shm(
@ -154,23 +175,58 @@ class Fsp:
return maybe_array return maybe_array
def fsp( @overload
wrapped=None, def fsp[**P](
wrapped: FspFunc[P],
*, *,
outputs: tuple[str] = (), outputs: tuple[str, ...] = (),
display_name: Optional[str] = None, display_name: str|None = None,
**config, **config: FspConfigValue,
) -> Fsp: ) -> Fsp[P]:
...
@overload
def fsp[**P](
wrapped: None = None,
*,
outputs: tuple[str, ...] = (),
display_name: str|None = None,
**config: FspConfigValue,
) -> Callable[[FspFunc[P]], Fsp[P]]:
...
def fsp[**P](
wrapped: FspFunc[P]|None = None,
*,
outputs: tuple[str, ...] = (),
display_name: str|None = None,
**config: FspConfigValue,
) -> Fsp[P]|Callable[[FspFunc[P]], Fsp[P]]:
'''
Wrap an FSP function directly or return its configured decorator.
Bare ``@fsp`` calls this function with ``wrapped`` and returns an
``Fsp``. Parameterized ``@fsp(...)`` returns ``decorate`` first;
Python then passes the decorated function to that closure.
'''
if wrapped is None: if wrapped is None:
return partial( def decorate(func: FspFunc[P]) -> Fsp[P]:
Fsp, return Fsp(
func,
outputs=outputs, outputs=outputs,
display_name=display_name, display_name=display_name,
**config, **config,
) )
return decorate
return Fsp(wrapped, outputs=(wrapped.__name__,)) return Fsp(wrapped, outputs=(wrapped.__name__,))
@ -180,7 +236,7 @@ def maybe_mk_fsp_shm(
size: int, size: int,
readonly: bool = True, readonly: bool = True,
) -> (str, ShmArray, bool): ) -> tuple[str, ShmArray, bool]:
''' '''
Allocate a single row shm array for an symbol-fsp pair if none Allocate a single row shm array for an symbol-fsp pair if none
exists, otherwise load the shm already existing for that token. exists, otherwise load the shm already existing for that token.

View File

@ -22,7 +22,10 @@ from __future__ import annotations
from contextlib import asynccontextmanager as acm from contextlib import asynccontextmanager as acm
from functools import partial from functools import partial
from typing import ( from typing import (
Any,
AsyncIterator, AsyncIterator,
Awaitable,
cast,
Callable, Callable,
TYPE_CHECKING, TYPE_CHECKING,
) )
@ -39,15 +42,20 @@ from ..log import (
get_console_log, get_console_log,
) )
from .. import data from .. import data
from ..data.ticktools import FeedQuote
from ..data.flows import Flume from ..data.flows import Flume
from ..data._sharedmem import NDTokenMsg
from tractor.ipc._shm import ShmArray from tractor.ipc._shm import ShmArray
from ..data._sampling import ( from ..data._sampling import (
_default_delay_s,
open_sample_stream, open_sample_stream,
) )
from ..accounting import MktPair from ..accounting import MktPair
from ._api import ( from ._api import (
Fsp, Fsp,
FspFunc,
FspStream,
FspYield,
FspHistory,
_load_builtins, _load_builtins,
NDToken, NDToken,
) )
@ -58,13 +66,70 @@ if TYPE_CHECKING:
log = get_logger(__name__) log = get_logger(__name__)
type QuoteFrame = dict[str, FeedQuote]
type BackfillNotice = tuple[str, float]|list[str|float]
type SampleValue = int|float|BackfillNotice
type SampleMsg = dict[str, SampleValue]
type EdgeFunc = FspFunc
def _is_relevant_sample_msg(
msg: SampleMsg,
fqme: str,
period_s: float,
) -> bool:
backfill: SampleValue|None = msg.get('backfilling')
if backfill is None:
return True
if (
not isinstance(backfill, (list, tuple))
or
len(backfill) != 2
):
return False
backfill_fqme: str | float
timeframe: str | float
backfill_fqme, timeframe = backfill
return (
backfill_fqme == fqme
and
timeframe == period_s
)
def _should_advance_dst(
msg: SampleMsg,
step_diff: int,
) -> bool:
return (
'backfilling' not in msg
and
step_diff == 1
)
def _needs_history_resync(
msg: SampleMsg,
synced: bool,
) -> bool:
return (
'backfilling' in msg
or
not synced
)
async def filter_quotes_by_sym( async def filter_quotes_by_sym(
sym: str, sym: str,
quote_stream: tractor.MsgStream, quote_stream: AsyncIterator[QuoteFrame],
) -> AsyncIterator[dict]: ) -> AsyncIterator[FeedQuote]:
''' '''
Filter quote stream by target symbol. Filter quote stream by target symbol.
@ -75,10 +140,91 @@ async def filter_quotes_by_sym(
yield {} yield {}
async for quotes in quote_stream: async for quotes in quote_stream:
quote = quotes.get(sym) quote: FeedQuote|None = quotes.get(sym)
if quote: if quote:
yield quote yield quote
class _HistoryShm:
'''
Present an immutable history view before switching to live SHM.
'''
def __init__(self, shm: ShmArray) -> None:
self._shm: ShmArray = shm
self.first: int = shm._first.value
self.last_index: int = shm._last.value
self.history: np.ndarray = shm._array[
self.first:self.last_index
].copy()
self.live: bool = False
@property
def array(self) -> np.ndarray:
if self.live:
return self._shm.array
return self.history
@property
def index(self) -> int:
if self.live:
return self._shm.index
return self.last_index % self._shm._len
def last(self, length: int = 1) -> np.ndarray:
return self.array[-length:]
def __getattr__(self, name: str) -> Any:
return getattr(self._shm, name)
async def _open_history_snapshot(
edge_func: EdgeFunc,
quote_stream: AsyncIterator[QuoteFrame],
fqme: str,
src_shm: ShmArray,
) -> tuple[
FspStream,
FspHistory,
tuple[int, int],
np.ndarray,
]:
'''
Start an FSP against an immutable source-history snapshot.
'''
history_shm: _HistoryShm = _HistoryShm(src_shm)
out_stream: FspStream = edge_func(
filter_quotes_by_sym(fqme, quote_stream),
cast(ShmArray, history_shm),
)
try:
first_yield: FspYield = await anext(out_stream)
except BaseException:
with trio.CancelScope(shield=True):
await out_stream.aclose()
raise
if isinstance(first_yield, tuple):
await out_stream.aclose()
raise TypeError(
f'FSP `{edge_func.__name__}` yielded a realtime update '
f'before its historical output'
)
history_output: FspHistory = first_yield
history_shm.live = True
return (
out_stream,
history_output,
(
history_shm.first,
history_shm.last_index,
),
history_shm.history,
)
# TODO: unifying the abstractions in this FSP subsys/layer: # TODO: unifying the abstractions in this FSP subsys/layer:
# -[ ] move the `.data.flows.Flume` type into this # -[ ] move the `.data.flows.Flume` type into this
# module/subsys/pkg? # module/subsys/pkg?
@ -124,24 +270,33 @@ class Cascade(Struct):
fsp: Fsp # UI-side middleware ctl API fsp: Fsp # UI-side middleware ctl API
# filled during cascade/.bind_func() (fsp_compute) init phases # filled during cascade/.bind_func() (fsp_compute) init phases
bind_func: Callable | None = None bind_func: Callable[..., Awaitable[None]]|None = None
complete: trio.Event | None = None complete: trio.Event|None = None
cs: trio.CancelScope | None = None cs: trio.CancelScope|None = None
client_stream: tractor.MsgStream | None = None client_stream: tractor.MsgStream|None = None
async def resync(self) -> int: async def resync(self) -> int:
# TODO: adopt an incremental update engine/approach # TODO: adopt an incremental update engine/approach
# where possible here eventually! # where possible here eventually!
log.info(f're-syncing fsp {self.fsp.name} to source') log.info(f're-syncing fsp {self.fsp.name} to source')
self.cs.cancel() cs: trio.CancelScope|None = self.cs
await self.complete.wait() complete: trio.Event|None = self.complete
index: int = await self.tn.start(self.bind_func) bind_func = self.bind_func
client_stream = self.client_stream
assert cs is not None
assert complete is not None
assert bind_func is not None
assert client_stream is not None
cs.cancel()
await complete.wait()
index: int = await self.tn.start(bind_func)
# always trigger UI refresh after history update, # always trigger UI refresh after history update,
# see ``piker.ui._fsp.FspAdmin.open_chain()`` and # see ``piker.ui._fsp.FspAdmin.open_chain()`` and
# ``piker.ui._display.trigger_update()``. # ``piker.ui._display.trigger_update()``.
dst_shm: ShmArray = self.dst.rt_shm dst_shm: ShmArray = self.dst.rt_shm
await self.client_stream.send({ await client_stream.send({
'fsp_update': { 'fsp_update': {
'key': dst_shm.token, 'key': dst_shm.token,
'first': dst_shm._first.value, 'first': dst_shm._first.value,
@ -158,27 +313,48 @@ class Cascade(Struct):
''' '''
src_shm: ShmArray = self.src.rt_shm src_shm: ShmArray = self.src.rt_shm
dst_shm: ShmArray = self.dst.rt_shm dst_shm: ShmArray = self.dst.rt_shm
step_diff = src_shm.index - dst_shm.index src_first = src_shm._first.value
len_diff = abs(len(src_shm.array) - len(dst_shm.array)) src_last = src_shm._last.value
synced: bool = not ( dst_first = dst_shm._first.value
# the source is likely backfilling and we must dst_last = dst_shm._last.value
# sync history calculations
len_diff > 2
# we aren't step synced to the source and may be # Compare absolute bounds instead of ``ShmArray.index``, which
# leading/lagging by a step # wraps at the allocation size. A one-step source lead is the
or step_diff > 1 # normal state before the destination sample is appended.
or step_diff < 0 step_diff = src_last - dst_last
len_diff = abs(
(src_last - src_first)
- (dst_last - dst_first)
)
synced: bool = (
src_first == dst_first
and
step_diff in (0, 1)
) )
if not synced: if not synced:
fsp: Fsp = self.fsp fsp: Fsp = self.fsp
report = (
f'ns-path: {fsp.ns_path!r}\n'
f'shm-token: {src_shm.token}\n'
f'src-bounds: {(src_first, src_last)}\n'
f'dst-bounds: {(dst_first, dst_last)}\n'
f'step_diff: {step_diff}\n'
f'len_diff: {len_diff}\n'
)
if (
src_first < dst_first
and
src_last == dst_last
):
log.info(
f'FSP source history grew:\n'
f'{report}'
)
else:
log.warning( log.warning(
f'***DESYNCED fsp***\n' f'***DESYNCED fsp***\n'
f'------------------\n' f'------------------\n'
f'ns-path: {fsp.ns_path!r}\n' f'{report}'
f'shm-token: {src_shm.token}\n'
f'step_diff: {step_diff}\n'
f'len_diff: {len_diff}\n'
) )
return ( return (
synced, synced,
@ -187,17 +363,20 @@ class Cascade(Struct):
) )
async def poll_and_sync_to_step(self) -> int: async def poll_and_sync_to_step(self) -> int:
synced, step_diff, _ = self.is_synced() while True:
while not synced:
await self.resync() await self.resync()
synced, step_diff, _ = self.is_synced() (
synced,
step_diff,
_,
) = self.is_synced()
if synced:
return step_diff return step_diff
@acm @acm
async def open_edge( async def open_edge(
self, self,
bind_func: Callable, bind_func: Callable[..., Awaitable[None]],
) -> int: ) -> int:
self.bind_func = bind_func self.bind_func = bind_func
index = await self.tn.start(bind_func) index = await self.tn.start(bind_func)
@ -206,14 +385,123 @@ class Cascade(Struct):
# -[ ] dynamic reconnection after update? # -[ ] dynamic reconnection after update?
async def _close_fsp_stream(out_stream: FspStream) -> None:
with trio.CancelScope(shield=True):
await out_stream.aclose()
def _write_history_output(
dst_shm: ShmArray,
history_output: FspHistory,
src_bounds: tuple[int, int],
src_history: np.ndarray,
func_name: str,
profiler: Profiler,
) -> int:
src_first, src_last = src_bounds
fields = dst_shm.array.dtype.fields
if fields is None:
raise TypeError('FSP destination SHM must be a structured array')
fields = fields.copy()
fields.pop('index')
fields.pop('time')
history_by_field: np.ndarray|None = None
if len(fields) > 1:
if not isinstance(history_output, dict):
raise ValueError(
f'`{func_name}` is a multi-output FSP and should '
f'yield a `dict[str, np.ndarray]` for history'
)
output_fields: list[str] = [
key
for key in fields
if key in history_output
]
if not output_fields:
raise ValueError(
f'`{func_name}` produced no declared history fields'
)
for key in output_fields:
output: np.ndarray|None = history_output[key]
if history_by_field is None:
length: int = (
len(src_history)
if output is None
else len(output)
)
# using the first output, determine the length of the
# struct-array that will be pushed to shm.
history_by_field = np.zeros(
length,
dtype=dst_shm.array.dtype,
)
if output is not None:
history_by_field[key] = output
else:
if not isinstance(history_output, np.ndarray):
raise ValueError(
f'`{func_name}` is a single output FSP and should '
f'yield an `np.ndarray` for history'
)
history_by_field = np.zeros(
len(history_output),
dtype=dst_shm.array.dtype,
)
history_by_field[func_name] = history_output
if history_by_field is None:
raise ValueError(f'`{func_name}` produced no history fields')
if len(history_by_field) != len(src_history):
raise ValueError(
f'`{func_name}` history length '
f'{len(history_by_field)} does not match source length '
f'{len(src_history)}'
)
history_by_field['time'] = src_history['time']
# TODO: XXX:
# THERE'S A BIG BUG HERE WITH THE `index` field since we're
# prepending a copy of the first value a few times to make
# sub-curves align with the parent bar chart.
# This likely needs to be fixed either by,
# - manually assigning the index and historical data
# seperately to the shm array (i.e. not using .push())
# - developing some system on top of the shared mem array that
# is `index` aware such that historical data can be indexed
# relative to the true first datum? Not sure if this is sane
# for incremental compuations.
first: int = src_first
dst_shm._first.value = first
# TODO: can we use this `start` flag instead of the manual
# setting above?
index: int = dst_shm.push(
history_by_field,
start=first,
)
profiler(f'{func_name} pushed history')
profiler.finish()
assert index == src_last
return index
async def connect_streams( async def connect_streams(
casc: Cascade, casc: Cascade,
mkt: MktPair, mkt: MktPair,
quote_stream: trio.abc.ReceiveChannel, quote_stream: AsyncIterator[QuoteFrame],
src: Flume, src: Flume,
dst: Flume, dst: Flume,
edge_func: Callable, edge_func: EdgeFunc,
# attach_stream: bool = False, # attach_stream: bool = False,
task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
@ -239,119 +527,38 @@ async def connect_streams(
# fqme: str = mkt.fqme # fqme: str = mkt.fqme
fqme: str = src.mkt.fqme fqme: str = src.mkt.fqme
# TODO: dynamic introspection of what the underlying (vertex)
# function actually requires from input node (flumes) then
# deliver those inputs as part of a graph "compilation" step?
out_stream = edge_func(
# TODO: do we even need this if we do the feed api right?
# shouldn't a local stream do this before we get a handle
# to the async iterable? it's that or we do some kinda
# async itertools style?
filter_quotes_by_sym(fqme, quote_stream),
# XXX: currently the ``ohlcv`` arg, but we should allow
# (dynamic) requests for src flume (node) streams?
src.rt_shm,
)
# HISTORY COMPUTE PHASE # HISTORY COMPUTE PHASE
# conduct a single iteration of fsp with historical bars input # conduct a single iteration of fsp with historical bars input
# and get historical output. # and get historical output.
history_output: ( history_output: FspHistory
dict[str, np.ndarray] # multi-output case src_shm: ShmArray = src.rt_shm
| np.ndarray, # single output case (
out_stream,
history_output,
src_bounds,
src_history,
) = await _open_history_snapshot(
edge_func,
quote_stream,
fqme,
src_shm,
) )
history_output = await anext(out_stream) func_name: str = edge_func.__name__
func_name = edge_func.__name__
profiler(f'{func_name} generated history') profiler(f'{func_name} generated history')
# build struct array with an 'index' field to push as history
# TODO: push using a[['f0', 'f1', .., 'fn']] = .. syntax no?
# if the output array is multi-field then push
# each respective field.
dst_shm: ShmArray = dst.rt_shm dst_shm: ShmArray = dst.rt_shm
fields = getattr(dst_shm.array.dtype, 'fields', None).copy() try:
fields.pop('index') index: int = _write_history_output(
history_by_field: np.ndarray | None = None dst_shm,
src_shm: ShmArray = src.rt_shm history_output,
src_time = src_shm.array['time'] src_bounds,
src_history,
if ( func_name,
fields and profiler,
len(fields) > 1
):
if not isinstance(history_output, dict):
raise ValueError(
f'`{func_name}` is a multi-output FSP and should yield a '
'`dict[str, np.ndarray]` for history'
) )
except BaseException:
for key in fields.keys(): await _close_fsp_stream(out_stream)
if key in history_output: raise
output = history_output[key]
if history_by_field is None:
if output is None:
length = len(src_shm.array)
else:
length = len(output)
# using the first output, determine
# the length of the struct-array that
# will be pushed to shm.
history_by_field = np.zeros(
length,
dtype=dst_shm.array.dtype
)
if output is None:
continue
history_by_field[key] = output
# single-key output stream
else:
if not isinstance(history_output, np.ndarray):
raise ValueError(
f'`{func_name}` is a single output FSP and should yield an '
'`np.ndarray` for history'
)
history_by_field = np.zeros(
len(history_output),
dtype=dst_shm.array.dtype
)
history_by_field[func_name] = history_output
history_by_field['time'] = src_time[-len(history_by_field):]
history_output['time'] = src_shm.array['time']
# TODO: XXX:
# THERE'S A BIG BUG HERE WITH THE `index` field since we're
# prepending a copy of the first value a few times to make
# sub-curves align with the parent bar chart.
# This likely needs to be fixed either by,
# - manually assigning the index and historical data
# seperately to the shm array (i.e. not using .push())
# - developing some system on top of the shared mem array that
# is `index` aware such that historical data can be indexed
# relative to the true first datum? Not sure if this is sane
# for incremental compuations.
first = dst_shm._first.value = src_shm._first.value
# TODO: can we use this `start` flag instead of the manual
# setting above?
index = dst_shm.push(
history_by_field,
start=first,
)
profiler(f'{func_name} pushed history')
profiler.finish()
# setup a respawn handle # setup a respawn handle
with trio.CancelScope() as cs: with trio.CancelScope() as cs:
@ -398,6 +605,9 @@ async def connect_streams(
# if hz > 60: # if hz > 60:
# log.info(f'FSP quote too fast: {hz}') # log.info(f'FSP quote too fast: {hz}')
# last = time.time() # last = time.time()
finally:
try:
await _close_fsp_stream(out_stream)
finally: finally:
casc.complete.set() casc.complete.set()
@ -410,11 +620,13 @@ async def cascade(
fqme: str, fqme: str,
# flume pair cascaded using an "edge function" # flume pair cascaded using an "edge function"
src_flume_addr: dict, src_flume_addr: dict[str, Any],
dst_flume_addr: dict, dst_flume_addr: dict[str, Any],
ns_path: NamespacePath, ns_path: NamespacePath,
shm_registry: dict[str, NDToken], shm_registry: list[
tuple[NDTokenMsg, str, NDTokenMsg]
],
zero_on_step: bool = False, zero_on_step: bool = False,
loglevel: str|None = None, loglevel: str|None = None,
@ -452,7 +664,7 @@ async def cascade(
# src: ShmArray = attach_shm_array(token=src_shm_token) # src: ShmArray = attach_shm_array(token=src_shm_token)
# dst: ShmArray = attach_shm_array(readonly=False, token=dst_shm_token) # dst: ShmArray = attach_shm_array(readonly=False, token=dst_shm_token)
reg = _load_builtins() reg: dict[NamespacePath, Fsp] = _load_builtins()
lines = '\n'.join([f'{key.rpartition(":")[2]} => {key}' for key in reg]) lines = '\n'.join([f'{key.rpartition(":")[2]} => {key}' for key in reg])
log.info( log.info(
f'Registered FSP set:\n{lines}' f'Registered FSP set:\n{lines}'
@ -465,18 +677,21 @@ async def cascade(
# not sure how else to do it. # not sure how else to do it.
for (token, fsp_name, dst_token) in shm_registry: for (token, fsp_name, dst_token) in shm_registry:
Fsp._flow_registry[( Fsp._flow_registry[(
NDToken.from_msg(token), NDToken.from_msg(dict(token)),
fsp_name, fsp_name,
)] = NDToken.from_msg(dst_token), None )] = NDToken.from_msg(dict(dst_token)), None
fsp: Fsp = reg.get( fsp: Fsp|None = reg.get(
NamespacePath(ns_path) NamespacePath(ns_path)
) )
func: Callable = fsp.func if (
fsp is None
if not func: or
not fsp.func
):
# TODO: assume it's a func target path # TODO: assume it's a func target path
raise ValueError(f'Unknown fsp target: {ns_path}') raise ValueError(f'Unknown fsp target: {ns_path}')
func: FspFunc = fsp.func
_fqme: str = src.mkt.fqme _fqme: str = src.mkt.fqme
assert _fqme == fqme assert _fqme == fqme
@ -521,7 +736,7 @@ async def cascade(
# the target task is spawned implicitly and then the event is # the target task is spawned implicitly and then the event is
# set via some higher level api? At that poing we might as well # set via some higher level api? At that poing we might as well
# be writing a one-cancels-one nursery though right? # be writing a one-cancels-one nursery though right?
casc = Cascade( casc: Cascade = Cascade(
src, src,
dst, dst,
tn, tn,
@ -529,7 +744,7 @@ async def cascade(
) )
# TODO: this seems like it should be wrapped somewhere? # TODO: this seems like it should be wrapped somewhere?
fsp_target = partial( fsp_target: Callable[..., Awaitable[None]] = partial(
connect_streams, connect_streams,
casc=casc, casc=casc,
mkt=mkt, mkt=mkt,
@ -543,9 +758,20 @@ async def cascade(
# and renders dst flume output(s) # and renders dst flume output(s)
edge_func=func edge_func=func
) )
async with casc.open_edge(
# Subscribe before history bootstrap so in-place repairs
# with unchanged SHM bounds can not lose invalidations.
period_s: float = src.rt_sample_period_s
async with (
open_sample_stream(
period_s=period_s,
loglevel=loglevel,
) as istream,
casc.open_edge(
bind_func=fsp_target, bind_func=fsp_target,
) as index: ) as index,
):
# casc.bind_func = fsp_target # casc.bind_func = fsp_target
# index = await tn.start(fsp_target) # index = await tn.start(fsp_target)
dst_shm: ShmArray = dst.rt_shm dst_shm: ShmArray = dst.rt_shm
@ -567,45 +793,51 @@ async def cascade(
async with ctx.open_stream() as client_stream: async with ctx.open_stream() as client_stream:
casc.client_stream: tractor.MsgStream = client_stream casc.client_stream: tractor.MsgStream = client_stream
s, step, ld = casc.is_synced()
# detect sample period step for subscription to increment
# signal
times = src.rt_shm.array['time']
if len(times) > 1:
last_ts = times[-1]
delay_s: float = float(last_ts - times[times != last_ts][-1])
else:
# our default "HFT" sample rate.
delay_s: float = _default_delay_s
# sub and increment the underlying shared memory buffer
# on every step msg received from the global `samplerd`
# service.
async with open_sample_stream(
period_s=float(delay_s),
loglevel=loglevel,
) as istream:
profiler(f'{func_name}: sample stream up') profiler(f'{func_name}: sample stream up')
profiler.finish() profiler.finish()
async for i in istream: async for sample_msg in istream:
# print(f'FSP incrementing {i}') # print(f'FSP incrementing {sample_msg}')
msg: SampleMsg = cast(
SampleMsg,
sample_msg,
)
# respawn the compute task if the source # ``samplerd`` broadcasts history events for
# array has been updated such that we compute # every feed and period. Only this cascade's
# new history from the (prepended) source. # source can invalidate its destination.
synced, step_diff, _ = casc.is_synced() if not _is_relevant_sample_msg(
if not synced: msg,
step_diff: int = await casc.poll_and_sync_to_step() fqme,
period_s,
# skip adding a last bar since we should already ):
# be step alinged continue
if step_diff == 0:
# Respawn the compute task after source history
# revisions or an actual bounds desync.
(
synced,
step_diff,
_,
) = casc.is_synced()
if _needs_history_resync(
msg,
synced,
):
step_diff = (
await casc.poll_and_sync_to_step()
)
# Backfill broadcasts only announce source
# history revisions. They never represent a
# new sample row. Duplicate sample wakeups are
# likewise non-advancing when already aligned.
if not _should_advance_dst(
msg,
step_diff,
):
continue continue
# read out last shm row, copy and write new row
array = dst_shm.array array = dst_shm.array
# some metrics like vlm should be reset # some metrics like vlm should be reset
@ -619,8 +851,12 @@ async def cascade(
# sync with source buffer's time step # sync with source buffer's time step
src_l2 = src_shm.array[-2:] src_l2 = src_shm.array[-2:]
src_li, src_lt = src_l2[-1][['index', 'time']] src_li, src_lt = src_l2[-1][
src_2li, src_2lt = src_l2[-2][['index', 'time']] ['index', 'time']
]
src_2li, src_2lt = src_l2[-2][
['index', 'time']
]
dst_shm._array['time'][src_li] = src_lt dst_shm._array['time'][src_li] = src_lt
dst_shm._array['time'][src_2li] = src_2lt dst_shm._array['time'][src_2li] = src_2lt

View File

@ -18,32 +18,37 @@
Momentum bby. Momentum bby.
""" """
from typing import AsyncIterator, Optional from typing import (
AsyncIterator,
)
import numpy as np import numpy as np
from numba import jit, float64, optional, int64 from numpy.typing import NDArray
from numba import jit
from ._api import fsp from ._api import (
from ..data import iterticks fsp,
FspStream,
)
from ..data import (
FeedQuote,
iterticks,
Tick,
)
from tractor.ipc._shm import ShmArray from tractor.ipc._shm import ShmArray
@jit( @jit(
float64[:](
float64[:],
optional(float64),
optional(float64)
),
nopython=True, nopython=True,
nogil=True nogil=True
) )
def ema( def ema(
y: 'np.ndarray[float64]', y: NDArray[np.float64],
alpha: optional(float64) = None, alpha: float|None = None,
ylast: optional(float64) = None, ylast: float|None = None,
) -> 'np.ndarray[float64]': ) -> NDArray[np.float64]:
r''' r'''
Exponential weighted moving average owka 'Exponential smoothing'. Exponential weighted moving average owka 'Exponential smoothing'.
@ -80,10 +85,14 @@ def ema(
# directly to the com of a SMA or WMA: # directly to the com of a SMA or WMA:
alpha = 2 / float(n + 1) alpha = 2 / float(n + 1)
s = np.empty(n, dtype=float64) s = np.empty(n, dtype=np.float64)
if n == 1: if n == 1:
s[0] = y[0] * alpha + ylast * (1 - alpha) s[0] = (
y[0]
if ylast is None
else y[0] * alpha + ylast * (1 - alpha)
)
else: else:
if ylast is None: if ylast is None:
@ -109,13 +118,12 @@ def ema(
# ) # )
def _rsi( def _rsi(
# TODO: use https://github.com/ramonhagenaars/nptyping signal: NDArray[np.float64],
signal: 'np.ndarray[float64]', period: int = 14,
period: int64 = 14, up_ema_last: float|None = None,
up_ema_last: float64 = None, down_ema_last: float|None = None,
down_ema_last: float64 = None,
) -> 'np.ndarray[float64]': ) -> tuple[NDArray[np.float64], float, float]:
''' '''
relative strengggth. relative strengggth.
@ -125,10 +133,18 @@ def _rsi(
df = np.diff(signal, prepend=0) df = np.diff(signal, prepend=0)
up = np.where(df > 0, df, 0) up = np.where(df > 0, df, 0)
up_ema = ema(up, alpha, up_ema_last) up_ema = ema(
up,
alpha,
up_ema_last,
)
down = np.where(df < 0, -df, 0) down = np.where(df < 0, -df, 0)
down_ema = ema(down, alpha, down_ema_last) down_ema = ema(
down,
alpha,
down_ema_last,
)
# avoid dbz errors, this leaves the first # avoid dbz errors, this leaves the first
# index == 0 right? # index == 0 right?
@ -151,7 +167,7 @@ def _wma(
signal: np.ndarray, signal: np.ndarray,
length: int, length: int,
weights: Optional[np.ndarray] = None, weights: np.ndarray|None = None,
) -> np.ndarray: ) -> np.ndarray:
''' '''
@ -174,11 +190,11 @@ def _wma(
@fsp @fsp
async def wma( async def wma(
source, #: AsyncStream[np.ndarray], source: AsyncIterator[FeedQuote],
length: int, ohlcv: ShmArray,
ohlcv: np.ndarray, # price time-frame "aware" length: int = 14,
) -> AsyncIterator[np.ndarray]: # maybe something like like FspStream? ) -> FspStream:
''' '''
Streaming weighted moving average. Streaming weighted moving average.
@ -188,23 +204,33 @@ async def wma(
''' '''
# deliver historical output as "first yield" # deliver historical output as "first yield"
yield _wma(ohlcv.array['close'], length) close: NDArray[np.float64] = ohlcv.array['close']
history: NDArray[np.float64] = np.full(len(close), np.nan)
if len(close) >= length:
history[length - 1:] = _wma(close, length)
yield history
# begin real-time section # begin real-time section
async for quote in source: async for quote in source:
for tick in iterticks(quote, type='trade'): _tick: Tick
yield _wma(ohlcv.last(length)) for _tick in iterticks(
quote,
types=['trade'],
):
closes: np.ndarray = ohlcv.last(length)['close']
if len(closes) == length:
yield 'wma', _wma(closes, length)[-1]
@fsp @fsp
async def rsi( async def rsi(
source: 'QuoteStream[Dict[str, Any]]', # noqa source: AsyncIterator[FeedQuote],
ohlcv: ShmArray, ohlcv: ShmArray,
period: int = 14, period: int = 14,
) -> AsyncIterator[np.ndarray]: ) -> FspStream:
''' '''
Multi-timeframe streaming RSI. Multi-timeframe streaming RSI.
@ -250,4 +276,4 @@ async def rsi(
up_ema_last=last_up_ema_close, up_ema_last=last_up_ema_close,
down_ema_last=last_down_ema_close, down_ema_last=last_down_ema_close,
) )
yield rsi_out[-1:] yield 'rsi', rsi_out[-1]

View File

@ -14,13 +14,17 @@
# You should have received a copy of the GNU Affero General Public License # You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
from typing import AsyncIterator, Optional, Union
import numpy as np import numpy as np
from tractor.trionics._broadcast import AsyncReceiver from tractor.trionics._broadcast import AsyncReceiver
from ._api import fsp from ._api import (
from ..data import iterticks fsp,
FspStream,
)
from ..data import (
FeedQuote,
iterticks,
)
from tractor.ipc._shm import ShmArray from tractor.ipc._shm import ShmArray
from ._momo import _wma from ._momo import _wma
from ..log import get_logger from ..log import get_logger
@ -36,7 +40,11 @@ def wap(
signal: np.ndarray, signal: np.ndarray,
weights: np.ndarray, weights: np.ndarray,
) -> np.ndarray: ) -> tuple[
np.ndarray, # weighted average price
np.ndarray, # cumulative weighted input
np.ndarray, # cumulative weights
]:
''' '''
Weighted average price from signal and weights. Weighted average price from signal and weights.
@ -62,16 +70,13 @@ def wap(
@fsp @fsp
async def tina_vwap( async def tina_vwap(
source: AsyncReceiver[dict], source: AsyncReceiver[FeedQuote],
ohlcv: ShmArray, # OHLC sampled history ohlcv: ShmArray, # OHLC sampled history
# TODO: anchor logic (eg. to session start) # TODO: anchor logic (eg. to session start)
anchors: Optional[np.ndarray] = None, anchors: np.ndarray|None = None,
) -> Union[ ) -> FspStream:
AsyncIterator[np.ndarray],
float
]:
''' '''
Streaming volume weighted moving average. Streaming volume weighted moving average.
@ -85,7 +90,7 @@ async def tina_vwap(
a = ohlcv.array a = ohlcv.array
chl3 = (a['close'] + a['high'] + a['low']) / 3 chl3 = (a['close'] + a['high'] + a['low']) / 3
v = a['volume'] v = np.maximum(a['volume'], 0)
h_vwap, cum_wp, cum_v = wap(chl3, v) h_vwap, cum_wp, cum_v = wap(chl3, v)
@ -107,7 +112,7 @@ async def tina_vwap(
# ] # ]
# this computes tick-by-tick weightings from here forward # this computes tick-by-tick weightings from here forward
size = tick['size'] size = max(float(tick['size']), 0.)
price = tick['price'] price = tick['price']
v_tot += size v_tot += size
@ -127,12 +132,10 @@ async def tina_vwap(
curve_style='step', curve_style='step',
) )
async def dolla_vlm( async def dolla_vlm(
source: AsyncReceiver[dict], source: AsyncReceiver[FeedQuote],
ohlcv: ShmArray, # OHLC sampled history ohlcv: ShmArray, # OHLC sampled history
) -> AsyncIterator[ ) -> FspStream:
tuple[str, Union[np.ndarray, float]],
]:
''' '''
"Dollar Volume", aka the volume in asset-currency-units (usually "Dollar Volume", aka the volume in asset-currency-units (usually
a fiat) computed from some price function for the sample step a fiat) computed from some price function for the sample step
@ -144,7 +147,7 @@ async def dolla_vlm(
''' '''
a = ohlcv.array a = ohlcv.array
chl3 = (a['close'] + a['high'] + a['low']) / 3 chl3 = (a['close'] + a['high'] + a['low']) / 3
v = a['volume'] v = np.maximum(a['volume'], 0)
# on first iteration yield history # on first iteration yield history
yield { yield {
@ -167,7 +170,7 @@ async def dolla_vlm(
): ):
# this computes tick-by-tick weightings from here forward # this computes tick-by-tick weightings from here forward
size = tick['size'] size = max(float(tick['size']), 0.)
price = tick['price'] price = tick['price']
li = ohlcv.index li = ohlcv.index
@ -227,14 +230,14 @@ async def dolla_vlm(
curve_style='line', curve_style='line',
) )
async def flow_rates( async def flow_rates(
source: AsyncReceiver[dict], source: AsyncReceiver[FeedQuote],
ohlcv: ShmArray, # OHLC sampled history ohlcv: ShmArray, # OHLC sampled history
# TODO (idea): a dynamic generic / boxing type that can be updated by other # TODO (idea): a dynamic generic / boxing type that can be updated by other
# FSPs, user input, and possibly any general event stream in # FSPs, user input, and possibly any general event stream in
# real-time. Hint: ideally implemented with caching until mutated # real-time. Hint: ideally implemented with caching until mutated
# ;) # ;)
period: 'Param[int]' = 1, # noqa period: int = 1,
# TODO: support other means by providing a map # TODO: support other means by providing a map
# to weights `partial()`-ed with `wma()`? # to weights `partial()`-ed with `wma()`?
@ -252,9 +255,7 @@ async def flow_rates(
# lazy copy in that case? # lazy copy in that case?
# dvlm: 'Fsp[dolla_vlm]' # dvlm: 'Fsp[dolla_vlm]'
) -> AsyncIterator[ ) -> FspStream:
tuple[str, Union[np.ndarray, float]],
]:
# generally no history available prior to real-time calcs # generally no history available prior to real-time calcs
yield { yield {
# from ib # from ib

View File

@ -71,6 +71,7 @@ from ..toolz.profile import (
if TYPE_CHECKING: if TYPE_CHECKING:
from . import _pg_overrides as pgo
from ._interaction import ChartView from ._interaction import ChartView
from ._chart import ChartPlotWidget from ._chart import ChartPlotWidget
from ._display import DisplayState from ._display import DisplayState
@ -282,7 +283,7 @@ class Viz(Struct):
''' '''
name: str name: str
plot: pg.PlotItem plot: pgo.PlotItem
_shm: ShmArray _shm: ShmArray
flume: Flume flume: Flume
graphics: Curve | BarItems graphics: Curve | BarItems
@ -318,6 +319,7 @@ class Viz(Struct):
_in_ds: bool = False _in_ds: bool = False
_index_step: float | None = None _index_step: float | None = None
_time_step: float | None = None _time_step: float | None = None
_last_fsp_update_sig: tuple[int, int, bytes]|None = None
# map from uppx -> (downsampled data, incremental graphics) # map from uppx -> (downsampled data, incremental graphics)
_src_r: Renderer | None = None _src_r: Renderer | None = None
@ -983,10 +985,10 @@ class Viz(Struct):
# XXX: SUPER UGGGHHH... without this we get stale cache # XXX: SUPER UGGGHHH... without this we get stale cache
# graphics that "smear" across the view horizontally # graphics that "smear" across the view horizontally
# when panning and the first datum is out of view.. # when panning and the first datum is out of view..
reset_cache = False
if ( if (
reset_cache reset_cache
or should_redraw or
should_redraw
): ):
# assign output paths to graphicis obj but # assign output paths to graphicis obj but
# after a coords-cache reset. # after a coords-cache reset.

View File

@ -20,6 +20,8 @@ FSP UI and graphics components.
Financial signal processing cluster and real-time graphics management. Financial signal processing cluster and real-time graphics management.
''' '''
from __future__ import annotations
from contextlib import asynccontextmanager as acm from contextlib import asynccontextmanager as acm
from functools import partial from functools import partial
import inspect import inspect
@ -27,9 +29,11 @@ from itertools import cycle
from typing import ( from typing import (
AsyncGenerator, AsyncGenerator,
Any, Any,
TYPE_CHECKING,
) )
import msgspec import msgspec
import numpy as np
import pyqtgraph as pg import pyqtgraph as pg
import tractor import tractor
from tractor.trionics import maybe_open_context from tractor.trionics import maybe_open_context
@ -53,7 +57,11 @@ from piker.data._sharedmem import try_read
from piker.log import get_logger from piker.log import get_logger
from piker.toolz import Profiler from piker.toolz import Profiler
from piker.types import Struct from piker.types import Struct
from ._axes import PriceAxis from ._axes import (
Axis,
PriceAxis,
YAxisLabel,
)
from ..calc import humanize from ..calc import humanize
from ._chart import ( from ._chart import (
ChartPlotWidget, ChartPlotWidget,
@ -65,26 +73,27 @@ from ._forms import (
open_form_input_handling, open_form_input_handling,
) )
if TYPE_CHECKING:
from ._dataviz import Viz
from ._interaction import ChartView
log = get_logger(__name__) log = get_logger(__name__)
def update_fsp_chart( def update_fsp_chart(
viz, viz: Viz,
graphics_name: str, graphics_name: str,
array_key: str|None, array_key: str,
**kwargs, force_redraw: bool = False,
) -> None: ) -> None:
shm = viz.shm shm: ShmArray = viz.shm
if not shm: array: np.ndarray = shm.array
return last_row: np.void|None = try_read(array)
array = shm.array
last_row = try_read(array)
# guard against unreadable case # guard against unreadable case
if not last_row: if last_row is None:
log.warning( log.warning(
f'Read-race on shm array,\n' f'Read-race on shm array,\n'
f'graphics_name: {graphics_name!r}\n' f'graphics_name: {graphics_name!r}\n'
@ -92,23 +101,100 @@ def update_fsp_chart(
) )
return return
# `update_fsp_chart()` runs for every source quote, while an FSP can
# publish more slowly. `Viz._last_fsp_update_sig` combines the
# destination's absolute half-open `ShmArray` bounds with bitwise
# `last_row['time']` and `last_row[array_key]` scalars. Comparing
# bytes keeps repeated NaNs stable; scalar equality treats every NaN
# as changed. A changed `ShmArray._first.value` catches history
# prepends, a changed `ShmArray._last.value` catches sample appends,
# and changed row bytes catch mutation of the current sample.
# `force_redraw` bypasses this gate because an equal-bound gap repair
# can rewrite earlier rows without changing the current row. This is
# only a repaint key, not synchronization proof; a mixed SHM read
# merely causes one extra later redraw.
update_sig: tuple[int, int, bytes] = (
shm._first.value,
shm._last.value,
np.asarray(last_row['time']).tobytes()
+ np.asarray(last_row[array_key]).tobytes(),
)
if (
not force_redraw
and
viz._last_fsp_update_sig == update_sig
):
return
# update graphics # update graphics
# NOTE: this does a length check internally which allows it # NOTE: this does a length check internally which allows it
# staying above the last row check below.. # staying above the last row check below..
viz.update_graphics() viz.update_graphics(force_redraw=force_redraw)
viz._last_fsp_update_sig = update_sig
# XXX: re: ``array_key``: fsp func names must be unique meaning we # XXX: re: ``array_key``: fsp func names must be unique meaning we
# can't have duplicates of the underlying data even if multiple # can't have duplicates of the underlying data even if multiple
# sub-charts reference it under different 'named charts'. # sub-charts reference it under different 'named charts'.
# read from last calculated value and update any label # read from last calculated value and update any label
last_val_sticky = viz.plot.getAxis( axis: Axis = viz.plot.getAxis('right')
'right')._stickies.get(graphics_name) last_val_sticky: YAxisLabel|None = axis._stickies.get(
if last_val_sticky: graphics_name,
last = last_row[array_key] )
if last_val_sticky is not None:
last: float = float(last_row[array_key])
last_val_sticky.update_from_data(-1, last) last_val_sticky.update_from_data(-1, last)
def update_fsp_vizs(
linked: LinkedSplits,
shm_token: NDToken,
) -> int:
'''
Redraw each `Viz` whose `Viz.shm` matches `shm_token`.
The FSP engine sends its destination token after rewriting
history. Every chart or subplot reading that SHM needs its
graphics caches updated and its view rescaled.
'''
charts: tuple[ChartPlotWidget, ...] = (
linked.chart,
*linked.subplots.values(),
)
updated: int = 0
views: list[ChartView] = []
for chart in charts:
for viz in chart._vizs.values():
viz: Viz
shm: ShmArray = viz.shm
if shm._token != shm_token:
continue
viz._mxmns.clear()
# Backfills can rewrite visible values without changing
# bounds, so bypass `Viz.update_graphics()` cache checks.
update_fsp_chart(
viz,
graphics_name=viz.name,
array_key=viz.name,
force_redraw=True,
)
updated += 1
view: ChartView = viz.plot.vb
if view not in views:
views.append(view)
for view in views:
view.interact_graphics_cycle(
do_linked_charts=False,
do_overlay_scaling=True,
)
return updated
@acm @acm
async def open_fsp_sidepane( async def open_fsp_sidepane(
linked: LinkedSplits, linked: LinkedSplits,
@ -285,7 +371,6 @@ async def run_fsp_ui(
# first UI update, usually from shm pushed history # first UI update, usually from shm pushed history
viz = chart.get_viz(array_key) viz = chart.get_viz(array_key)
update_fsp_chart( update_fsp_chart(
chart,
viz, viz,
name, name,
array_key=array_key, array_key=array_key,
@ -461,9 +546,11 @@ class FspAdmin:
log.debug( log.debug(
f'Re-syncing graphics for fsp: {ns_path}' f'Re-syncing graphics for fsp: {ns_path}'
) )
self.linked.graphics_cycle( update_fsp_vizs(
trigger_all=True, self.linked,
prepend_update_index=info['first'], shm_token=NDToken.from_msg(
dict(info['key']),
),
) )
else: else:
log.info(f'recved unexpected fsp engine msg: {msg}') log.info(f'recved unexpected fsp engine msg: {msg}')

View File

@ -22,8 +22,20 @@ Generally, our does not require "scentific precision" for pixel perfect
view transforms. view transforms.
""" """
from __future__ import annotations
from typing import (
Callable,
TYPE_CHECKING,
)
import pyqtgraph as pg import pyqtgraph as pg
if TYPE_CHECKING:
from ._axes import Axis
from ._chart import ChartPlotWidget
from ._interaction import ChartView
def invertQTransform(tr): def invertQTransform(tr):
"""Return a QTransform that is the inverse of *tr*. """Return a QTransform that is the inverse of *tr*.
@ -83,6 +95,11 @@ class PlotItem(pg.PlotItem):
options options
''' '''
vb: ChartView
chart_widget: ChartPlotWidget|None
if TYPE_CHECKING:
getAxis: Callable[[str], Axis]
def __init__( def __init__(
self, self,
parent=None, parent=None,

View File

@ -5,20 +5,26 @@
- config file IO - config file IO
''' '''
from decimal import Decimal
from pathlib import Path from pathlib import Path
from types import ModuleType
import tomllib
import pytest import pytest
from piker import config from piker import config
from piker.accounting import ( from piker.accounting import (
Account, Account,
Asset,
calc, calc,
open_account, open_account,
load_account, load_account,
load_account_from_ledger, load_account_from_ledger,
MktPair,
open_trade_ledger, open_trade_ledger,
Position, Position,
TransactionLedger, TransactionLedger,
) )
from piker.data._symcache import SymbologyCache
import tractor import tractor
@ -47,6 +53,82 @@ def test_account_file_default_empty(
assert path.parent.name == 'accounting' assert path.parent.name == 'accounting'
def test_paper_ledger_fuzzy_qualifies_mktmap_keys(
tmp_path: Path,
):
'''
Qualify legacy paper-ledger FQMEs via symcache string keys.
``SymbologyCache.search()`` previously passed its ``mktmaps``
mapping directly to RapidFuzz. RapidFuzz searches mapping values,
so a paper fill with an unqualified FQME made RapidFuzz call
``len()`` on a ``MktPair`` while
``TransactionLedger.write_config()`` exited.
This test installs both native-ID and FQME keys for one market,
proves search preserves the matched alias, then writes the same
unqualified paper transaction shape. The persisted fully
qualified keys prove the ledger exit path resolves the market
without treating ``MktPair`` values as fuzzy-search sequences.
'''
usdt = Asset(
name='usdt',
atype='crypto',
tx_tick=Decimal('0.00000001'),
)
nvda = Asset(
name='nvda',
atype='stock',
tx_tick=Decimal('0.00000001'),
)
mkt = MktPair(
dst=nvda,
src=usdt,
price_tick=Decimal('0.01'),
size_tick=Decimal('0.001'),
bs_mktid='NVDAUSDT',
broker='binance',
venue='usdtm',
expiry='perp',
)
symcache = SymbologyCache(
mod=ModuleType('binance'),
fp=tmp_path / 'binance.symcache.toml',
mktmaps={
mkt.bs_mktid: mkt,
mkt.fqme: mkt,
},
)
matches = symcache.search(mkt.bs_mktid)
assert next(iter(matches)) == mkt.bs_mktid
assert matches[mkt.bs_mktid] is mkt
ledger_path = tmp_path / 'trades_binance_paper.toml'
unqualified_fqme = 'nvdausdt.usdtm.perp'
ledger = TransactionLedger(
ledger_dict={
'fill-id': {
'fqme': unqualified_fqme,
'bs_mktid': unqualified_fqme,
},
},
file_path=ledger_path,
account='paper',
mod=symcache.mod,
tx_sort=lambda txns: txns.items(),
symcache=symcache,
)
ledger.write_config()
with ledger_path.open('rb') as ledger_file:
saved = tomllib.load(ledger_file)['fill-id']
assert saved['fqme'] == mkt.fqme
assert saved['bs_mktid'] == mkt.fqme
@pytest.mark.parametrize( @pytest.mark.parametrize(
'fq_acnt', 'fq_acnt',
[ [

View File

@ -0,0 +1,64 @@
'''
Momentum FSP numerical regressions.
'''
import numpy as np
import pytest
from piker.fsp._momo import ema
@pytest.mark.parametrize(
'args,kwargs,expected',
[
((), {}, [1., 1.5, 2.25]),
((None,), {}, [1., 1.5, 2.25]),
((None, None), {}, [1., 1.5, 2.25]),
((0.25,), {}, [1., 1.25, 1.6875]),
((), {'alpha': 0.25}, [1., 1.25, 1.6875]),
((), {'ylast': 4.}, [4., 3., 3.]),
((None, 4.), {}, [4., 3., 3.]),
((0.5, 4.), {}, [4., 3., 3.]),
],
)
def test_ema_optional_arguments(
args: tuple[float|None, ...],
kwargs: dict[str, float],
expected: list[float],
) -> None:
'''
Accept omitted EMA defaults through the real Numba dispatcher.
`ema()` exposed optional smoothing and seed arguments, but its
eager signature accepted only explicit values or `None`.
Numba's omitted-argument types therefore raised `TypeError`
before the numerical kernel ran. Exercise positional and keyword
omissions as well as explicit arguments, checking the resulting
recurrence against hand-computed values. Import the production
dispatcher so this catches signature regressions that a call to
`ema.py_func` would miss.
'''
signal: np.ndarray = np.array([1., 2., 3.])
result: np.ndarray = ema(signal, *args, **kwargs)
np.testing.assert_allclose(result, expected)
assert result.dtype == np.float64
def test_ema_single_sample_continuation() -> None:
'''
Preserve the previous EMA when advancing one realtime sample.
The old one-sample path multiplied an absent seed by a float,
failing instead of initializing from the sample. Removing the
eager Numba signature must retain the existing initialization
fix and the previous-value update used by realtime RSI. Use a
distinct seed and smoothing factor so copying either the seed
or sample fails; verify an omitted seed uses the sole sample.
'''
signal: np.ndarray = np.array([3.])
np.testing.assert_allclose(ema(signal), [3.])
np.testing.assert_allclose(ema(signal, 0.25, 7.), [6.])

View File

@ -0,0 +1,610 @@
'''
FSP history synchronization regressions.
'''
from collections.abc import AsyncIterator
from types import SimpleNamespace
from typing import cast
import numpy as np
import pytest
import trio
from piker.fsp._engine import (
_is_relevant_sample_msg,
_needs_history_resync,
_open_history_snapshot,
_should_advance_dst,
Cascade,
EdgeFunc,
QuoteFrame,
)
from piker.fsp._momo import (
rsi,
wma,
)
from piker.fsp._volume import (
dolla_vlm,
tina_vwap,
)
from piker.accounting import MktPair
from piker.data.flows import Flume
from piker.data._sharedmem import NDTokenMsg
from piker.data.ticktools import FeedQuote
from piker.fsp._api import Fsp
from piker.ui._chart import LinkedSplits
from piker.ui._fsp import (
update_fsp_chart,
update_fsp_vizs,
)
from tractor.ipc._shm import (
NDToken,
ShmArray,
)
type SampleMsg = dict[
str,
int|float|tuple[str, float],
]
def test_sample_period_ignores_market_closure_gap() -> None:
'''
Keep FSP cascades subscribed to their regular sampler period.
Starting a cascade immediately after a market closure left the
source SHM tail with a large gap between its last two distinct
timestamps. Using only that pair subscribed the cascade to the gap
duration instead of the regular one-second stream. Realtime FSP
writes then repeatedly replaced one row without advancing its SHM
bound. Give both source SHMs closure-gap tail deltas and prove
`Flume.get_ds_info()` returns its declared one- and 60-second
periods instead of deriving cadence from those rows.
'''
rt_shm = OhlcvShm(length=4)
hist_shm = OhlcvShm(length=4)
rt_shm._array['time'][:4] = [100, 101, 102, 3600]
hist_shm._array['time'][:4] = [100, 160, 220, 3600]
flume = Flume(
mkt=cast(MktPair, SimpleNamespace()),
first_quote={},
_rt_shm_token=rt_shm._token,
_hist_shm_token=hist_shm._token,
)
flume._rt_shm = cast(ShmArray, rt_shm)
flume._hist_shm = cast(ShmArray, hist_shm)
assert np.diff(rt_shm.array['time'])[-1] == 3498
assert np.diff(hist_shm.array['time'])[-1] == 3380
assert flume.get_ds_info() == (1, 60, 60)
class Value:
def __init__(self, value: int) -> None:
self.value: int = value
class Shm:
def __init__(
self,
first: int,
last: int,
token: str = 'fsp',
) -> None:
self._first: Value = Value(first)
self._last: Value = Value(last)
self._array: np.ndarray = np.ones(4096)
self._len: int = len(self._array)
self._token: NDToken = NDToken(
shm_name=token,
shm_first_index_name=f'{token}_first',
shm_last_index_name=f'{token}_last',
dtype_descr=(('value', '<f8'),),
size=len(self._array),
)
@property
def array(self) -> np.ndarray:
return self._array[
self._first.value:self._last.value
]
@property
def token(self) -> NDTokenMsg:
return cast(NDTokenMsg, self._token.as_msg())
@property
def index(self) -> int:
return self._last.value % len(self._array)
def last(self, length: int = 1) -> np.ndarray:
return self.array[-length:]
class OhlcvShm(Shm):
def __init__(self, length: int = 32) -> None:
dtype = np.dtype([
('index', '<i8'),
('time', '<i8'),
('open', '<f8'),
('high', '<f8'),
('low', '<f8'),
('close', '<f8'),
('volume', '<f8'),
])
self._first = Value(0)
self._last = Value(length)
self._array = np.ones(length + 8, dtype=dtype)
self._array['index'] = np.arange(length + 8)
self._array['time'] = np.arange(length + 8)
self._array['close'] = np.arange(length + 8) + 1
self._array['high'] = self._array['close'] + 1
self._array['low'] = self._array['close'] - 1
self._len = len(self._array)
self._token = NDToken(
shm_name='ohlcv',
shm_first_index_name='ohlcv_first',
shm_last_index_name='ohlcv_last',
dtype_descr=tuple(dtype.descr),
size=self._len,
)
class FspShm(Shm):
def __init__(
self,
first: int = 0,
last: int = 4,
token: str = 'fsp',
) -> None:
dtype = np.dtype([
('index', '<i8'),
('time', '<f8'),
('flow', '<f8'),
('dark_flow', '<f8'),
])
size: int = max(last + 2, 4096)
self._first = Value(first)
self._last = Value(last)
self._array = np.zeros(size, dtype=dtype)
self._array['index'] = np.arange(size)
self._array['time'] = np.arange(size)
self._array['flow'] = np.arange(size)
self._len = len(self._array)
self._token = NDToken(
shm_name=token,
shm_first_index_name=f'{token}_first',
shm_last_index_name=f'{token}_last',
dtype_descr=tuple(dtype.descr),
size=self._len,
)
def mk_cascade(
src_bounds: tuple[int, int],
dst_bounds: tuple[int, int],
) -> Cascade:
fsp: Fsp = cast(
Fsp,
SimpleNamespace(
name='test_fsp',
ns_path='tests:test_fsp',
),
)
src: Flume = cast(
Flume,
SimpleNamespace(
rt_shm=Shm(*src_bounds, token='src'),
),
)
dst: Flume = cast(
Flume,
SimpleNamespace(
rt_shm=Shm(*dst_bounds, token='dst'),
),
)
nursery: trio.Nursery = cast(trio.Nursery, None)
return Cascade(src, dst, nursery, fsp)
@pytest.mark.parametrize(
'src_bounds,dst_bounds,expected',
[
((10, 20), (10, 20), (True, 0, 0)),
((10, 21), (10, 20), (True, 1, 1)),
((0, 3000), (2000, 3000), (False, 0, 2000)),
((9, 20), (10, 20), (False, 0, 1)),
((8, 20), (10, 20), (False, 0, 2)),
((10, 20), (9, 20), (False, 0, 1)),
((11, 21), (10, 20), (False, 1, 0)),
((10, 22), (10, 20), (False, 2, 2)),
((10, 19), (10, 20), (False, -1, 1)),
],
)
def test_cascade_sync_uses_absolute_bounds(
src_bounds: tuple[int, int],
dst_bounds: tuple[int, int],
expected: tuple[bool, int, int],
) -> None:
'''
Detect every history-bound mismatch without modulo-index skew.
Source prepends move only its first bound, while a normal realtime
step moves only its last bound. The former requires a historical
recompute even for one or two rows; the latter permits exactly one
destination append. This matrix also shifts equal-length bounds and
puts the destination ahead to prove neither state is accepted.
'''
cascade = mk_cascade(src_bounds, dst_bounds)
assert cascade.is_synced() == expected
@pytest.mark.parametrize(
'msg,expected',
[
({'index': 1}, True),
({'backfilling': ('tsla.nasdaq.ib', 1)}, True),
({'backfilling': ('btcusdt.binance', 1)}, False),
({'backfilling': ('tsla.nasdaq.ib', 60)}, False),
],
)
def test_cascade_filters_foreign_backfill_events(
msg: SampleMsg,
expected: bool,
) -> None:
'''
Ignore backfill wakeups from overlay markets and other periods.
Samplerd broadcasts every backfill marker to every period and FSP
subscriber. With multiple markets overlaid, an unrelated provider
frame used to wake each primary-market cascade and race its own SHM
inspection. Feed regular sample steps and matching history events
through, but reject both a foreign FQME and the 60-second period.
'''
assert _is_relevant_sample_msg(
msg,
fqme='tsla.nasdaq.ib',
period_s=1,
) is expected
@pytest.mark.parametrize(
'msg,step_diff,expected',
[
({'index': 1}, 1, True),
({'index': 1}, 0, False),
({'backfilling': ('tsla.nasdaq.ib', 1)}, 1, False),
({'backfilling': ('tsla.nasdaq.ib', 1)}, 0, False),
],
)
def test_only_new_sample_steps_advance_destination(
msg: SampleMsg,
step_diff: int,
expected: bool,
) -> None:
'''
Keep queued history and duplicate wakeups from adding FSP rows.
A recomputation can consume several prepends before their sampler
markers are delivered. Those stale markers then observe aligned SHM
bounds and previously appended duplicate destination rows. Model the
queued marker and regular duplicate cases, and prove only a genuine
one-step source lead authorizes destination advancement.
'''
assert _should_advance_dst(msg, step_diff) is expected
@pytest.mark.parametrize(
'msg,synced,expected',
[
({'index': 1}, True, False),
({'index': 1}, False, True),
({'backfilling': ('tsla.nasdaq.ib', 1)}, True, True),
({'backfilling': ('tsla.nasdaq.ib', 1)}, False, True),
],
)
def test_backfill_always_invalidates_fsp_history(
msg: SampleMsg,
synced: bool,
expected: bool,
) -> None:
'''
Recompute in-place history repairs with unchanged SHM bounds.
Gap repair can replace null source rows without moving either SHM
bound. A bounds-only predicate therefore reports synchronization
even though derived values are stale. Prove every relevant backfill
marker forces history replay while an aligned sample event does not.
'''
assert _needs_history_resync(msg, synced) is expected
def test_history_compute_uses_immutable_source_snapshot() -> None:
'''
Do not publish output and timestamps from different source ranges.
A source append or prepend can occur while the FSP's first yield is
computing. The old path read timestamps only afterward and could pair
an N-row result with a shifted N-row timestamp tail. Mutate the live
last bound during the first yield, then prove the accepted output and
timestamps retain the original range while later reads switch to live
SHM.
'''
shm = Shm(10, 20, token='source')
live_lengths: list[int] = []
async def edge(
_source: AsyncIterator[FeedQuote],
src_shm: ShmArray,
) -> AsyncIterator[np.ndarray]:
shm._last.value += 1
yield np.ones(len(src_shm.array))
live_lengths.append(len(src_shm.array))
async def main() -> None:
(
out_stream,
history,
bounds,
src_history,
) = await _open_history_snapshot(
cast(EdgeFunc, edge),
quote_stream=cast(
AsyncIterator[QuoteFrame],
object(),
),
fqme='tsla.nasdaq.ib',
src_shm=cast(ShmArray, shm),
)
assert bounds == (10, 20)
assert len(history) == 10
assert len(src_history) == 10
with pytest.raises(StopAsyncIteration):
await anext(out_stream)
trio.run(main)
assert live_lengths == [11]
@pytest.mark.parametrize('target', [wma, rsi, tina_vwap])
def test_builtin_fsp_stream_contract(target: Fsp) -> None:
'''
Keep every registered scalar FSP on the engine's yield protocol.
The momentum operators previously had incompatible call signatures,
short historical arrays, and bare realtime yields, all hidden by an
engine-side cast. Run each against one OHLCV snapshot and one trade,
proving the first yield is a source-aligned array and the next yield
is a named realtime field/value pair.
'''
shm = cast(ShmArray, OhlcvShm())
async def source() -> AsyncIterator[FeedQuote]:
yield {
'ticks': [{
'type': 'trade',
'price': 42.0,
'size': 1.0,
}],
}
async def main() -> None:
stream = target.func(source(), shm)
history = await anext(stream)
assert isinstance(history, np.ndarray)
assert len(history) == len(shm.array)
realtime = await anext(stream)
assert isinstance(realtime, tuple)
assert realtime[0] == target.name
await stream.aclose()
trio.run(main)
def test_dolla_vlm_rejects_negative_trade_volume() -> None:
'''
Keep a provider volume reset from hiding the zoomed-out curve.
IB can emit a large negative trade size around a session closure.
The sampler previously stored that size as bar volume and the
dollar-volume history multiplied it by price. The resulting outlier
made `Viz.maxmin()` reject the entire downsampled range, leaving a
stale cached path or no visible curve. Seed one historical negative
volume and stream one negative trade tick, then prove both historical
and realtime dollar volume remain nonnegative.
'''
shm = OhlcvShm(length=4)
shm._array['volume'][:4] = [1, 2, -2_319_096, 3]
async def source() -> AsyncIterator[FeedQuote]:
yield {
'ticks': [{
'type': 'trade',
'price': 29_495.75,
'size': -2_319_096.,
}],
}
async def main() -> None:
stream = dolla_vlm.func(
source(),
cast(ShmArray, shm),
)
history = await anext(stream)
assert isinstance(history, dict)
values = history['dolla_vlm']
assert isinstance(values, np.ndarray)
assert values[2] == 0
assert np.all(values >= 0)
realtime = await anext(stream)
assert realtime == ('dolla_vlm', 0)
await stream.aclose()
trio.run(main)
class Viz:
def __init__(self, name: str, shm: Shm) -> None:
self.name: str = name
self.shm: Shm = shm
self.updates: int = 0
self.force_redraws: list[bool] = []
self._last_fsp_update_sig: (
tuple[int, int, bytes]|None
) = None
self._mxmns: dict[
tuple[int, int],
tuple[int, int],
] = {(10, 20): (1, 2)}
self.view = SimpleNamespace(
rescale_count=0,
)
def rescale(
*,
do_linked_charts: bool,
do_overlay_scaling: bool,
) -> None:
assert not do_linked_charts
assert do_overlay_scaling
self.view.rescale_count += 1
self.view.interact_graphics_cycle = rescale
self.plot: SimpleNamespace = SimpleNamespace(
getAxis=lambda name: SimpleNamespace(_stickies={}),
vb=self.view,
)
def update_graphics(
self,
force_redraw: bool = False,
) -> None:
self.updates += 1
self.force_redraws.append(force_redraw)
def test_fsp_chart_skips_unchanged_realtime_state() -> None:
'''
Avoid repainting an unchanged current FSP datum on every quote.
The display loop calls `update_fsp_chart()` for every realtime
quote, even when a given FSP destination has not advanced or
changed its current value. That rebuilt and painted the same last
step rectangle repeatedly. Exercise identical state, an in-place
value update, a new sample with the same value, and a forced
history repair. The update counts prove unchanged requests are
skipped while each legitimate invalidation still redraws. A final
stable-NaN pair proves scalar NaN inequality can not defeat the
guard.
'''
shm = FspShm()
viz = Viz('flow', shm)
update_fsp_chart(viz, 'flow', 'flow')
update_fsp_chart(viz, 'flow', 'flow')
assert viz.updates == 1
shm._array['flow'][shm._last.value - 1] = 42
update_fsp_chart(viz, 'flow', 'flow')
assert viz.updates == 2
shm._array['flow'][shm._last.value] = 42
shm._last.value += 1
update_fsp_chart(viz, 'flow', 'flow')
assert viz.updates == 3
update_fsp_chart(
viz,
'flow',
'flow',
force_redraw=True,
)
assert viz.updates == 4
shm._array['flow'][shm._last.value - 1] = np.nan
update_fsp_chart(viz, 'flow', 'flow')
update_fsp_chart(viz, 'flow', 'flow')
assert viz.updates == 5
assert viz.force_redraws == [
False,
False,
False,
True,
False,
]
def test_fsp_history_update_redraws_only_derived_vizs() -> None:
'''
Keep a source-history prepend from refreshing the whole chart.
FSP cascades recompute after each near-term backfill frame. The old
notification called the linked chart's full graphics cycle, which
redrew the primary market and every overlay and disturbed the live
view repeatedly. Arrange two visualizations sharing the rebuilt FSP
SHM plus unrelated source and overlay SHMs, then prove only the two
derived curves receive an update.
'''
fsp_shm = FspShm(10, 20, token='derived')
source_viz = Viz('source', Shm(10, 20, token='source'))
first_fsp_viz = Viz('flow', fsp_shm)
second_fsp_viz = Viz('dark_flow', fsp_shm)
overlay_viz = Viz('overlay', Shm(10, 20, token='overlay'))
linked: LinkedSplits = cast(
LinkedSplits,
SimpleNamespace(
chart=SimpleNamespace(
_vizs={
'source': source_viz,
'overlay': overlay_viz,
},
),
subplots={
'volume': SimpleNamespace(
_vizs={
'flow': first_fsp_viz,
'dark_flow': second_fsp_viz,
},
),
},
),
)
updated = update_fsp_vizs(linked, fsp_shm._token)
assert updated == 2
assert first_fsp_viz.updates == 1
assert second_fsp_viz.updates == 1
assert first_fsp_viz.force_redraws == [True]
assert second_fsp_viz.force_redraws == [True]
assert first_fsp_viz._mxmns == {}
assert second_fsp_viz._mxmns == {}
assert first_fsp_viz.view.rescale_count == 1
assert second_fsp_viz.view.rescale_count == 1
assert source_viz.updates == 0
assert overlay_viz.updates == 0

View File

@ -0,0 +1,99 @@
'''
IB broker event-relay regressions.
'''
from collections.abc import AsyncIterator
from typing import Any
import pytest
import trio
from piker.brokers.ib.broker import deliver_trade_events
from piker.clearing import OrderDialogs
from piker.clearing._messages import BrokerdError
@pytest.mark.parametrize(
('error', 'expected_reason'),
[
(
'[code 104] connection failed',
' connection failed',
),
(
'[Errno 111] Connect call failed',
' Connect call failed',
),
(
'plain API error',
'plain API error',
),
],
)
def test_trade_error_strings_are_relayed(
error: str,
expected_reason: str,
) -> None:
'''
Raw IB API error strings must not kill the trade-event relay.
``Client.inline_errors()`` can emit bracketed ``[code N]`` and
``[Errno N]`` strings instead of structured error dictionaries.
The string branch called ``str.partition()`` without a separator,
raising ``TypeError`` before EMS received the error and
terminating all later trade-event delivery. Feed both documented
bracket forms and an unprefixed control through
``deliver_trade_events()``. The exact emitted reason and unknown
request ID prove each event crossed the real relay path without
crashing or being misclassified.
'''
class FakeEmsStream:
'''
Collect messages sent to the EMS stream.
'''
def __init__(self) -> None:
'''
Initialize the captured message sequence.
'''
self.messages: list[Any] = []
async def send(self, msg: Any) -> None:
'''
Capture one relayed broker message.
'''
self.messages.append(msg)
async def events() -> AsyncIterator[tuple[str, str]]:
'''
Emit the raw IB error under test.
'''
yield 'error', error
async def main() -> list[Any]:
'''
Drive the broker relay to input exhaustion.
'''
ems_stream = FakeEmsStream()
await deliver_trade_events(
events(),
ems_stream,
accounts_def={},
proxies={},
ledgers={},
tables={},
flows=OrderDialogs(_flows={}),
)
return ems_stream.messages
messages = trio.run(main)
assert len(messages) == 1
msg = messages[0]
assert isinstance(msg, BrokerdError)
assert msg.reqid == '<unknown>'
assert msg.reason == expected_reason