Compare commits
7 Commits
14e6a319ed
...
f15f8178a3
| Author | SHA1 | Date |
|---|---|---|
|
|
f15f8178a3 | |
|
|
eee19de090 | |
|
|
59d5d9a66d | |
|
|
6418121923 | |
|
|
119d2c0495 | |
|
|
bc6e18d7b4 | |
|
|
4485f2b9ce |
|
|
@ -0,0 +1,59 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:08:59Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T170859Z_75cefe10_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Session-initiating instruction (driving all 7 commits in
|
||||||
|
this series):
|
||||||
|
|
||||||
|
> ok i want you to become the distributed runtime and
|
||||||
|
> concurrency expert for this project - namely acquire
|
||||||
|
> a deep understanding of tractor and how it's used.
|
||||||
|
> then i want you to attempt to factor our current
|
||||||
|
> brokerd service daemon into 2 daemons: a brokerd
|
||||||
|
> which only hosts live/paper trading endpoint tasks
|
||||||
|
> [..] a new `datad` subdaemon which in a separate
|
||||||
|
> subactor serves all the data feed service tasks [..]
|
||||||
|
> give me a mega detailed plan on how to approach this,
|
||||||
|
> and a staged approach for the implementation.
|
||||||
|
|
||||||
|
Proximate driver for THIS commit: during the approved
|
||||||
|
plan's stage-0 test gate the agent discovered test
|
||||||
|
subactors were writing into the user's REAL
|
||||||
|
`~/.config/piker/accounting/` files (a bogus paper fill
|
||||||
|
landed in `account.kraken.paper.toml`) because the old
|
||||||
|
test-dir override in `config.get_app_dir()` was
|
||||||
|
commented out and could never work in spawned
|
||||||
|
subactors anyway. Fix applied autonomously under the
|
||||||
|
approved plan; no explicit per-fix user prompt.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Restore `pytest` config-dir isolation by resolving the
|
||||||
|
per-test tmp dir lazily (at conf-path access time) from
|
||||||
|
`tractor` runtime-vars, which propagate down the actor
|
||||||
|
tree; route all conf-path derivation through
|
||||||
|
`config.get_conf_dir()` so the override is effective in
|
||||||
|
every (sub)actor.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/config.py` — add `_maybe_use_test_dir()`;
|
||||||
|
hook in `get_conf_dir()`; route `get_conf_path()` +
|
||||||
|
`load()` mkdir through it
|
||||||
|
- `piker/accounting/_ledger.py` — derive ledger dir
|
||||||
|
via `config.get_conf_dir()` (not the global)
|
||||||
|
- `piker/accounting/_pos.py` — same for account files
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,46 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:08:59Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/config.py`
|
||||||
|
|
||||||
|
Generated: `config._maybe_use_test_dir()` — lazily
|
||||||
|
reads `piker_test_dir` from
|
||||||
|
`tractor.runtime._state._runtime_vars['piker_vars']`
|
||||||
|
(pre-loaded by `open_piker_runtime()` from the
|
||||||
|
`tests.conftest._open_test_pikerd()` overrides) and
|
||||||
|
calls `_override_config_dir()` when set. Hooked at the
|
||||||
|
top of `get_conf_dir()`; `get_conf_path()` and
|
||||||
|
`load()`'s dir-creation rerouted through
|
||||||
|
`get_conf_dir()`.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/accounting/_ledger.py`
|
||||||
|
> `git log -1 -p --follow -- piker/accounting/_pos.py`
|
||||||
|
|
||||||
|
Generated: ledger/account dir derivation switched from
|
||||||
|
the `config._config_dir` module global to
|
||||||
|
`config.get_conf_dir()` + `mkdir(parents=True,
|
||||||
|
exist_ok=True)` for nested tmp-dir creation.
|
||||||
|
|
||||||
|
Key diagnostic reasoning (verbatim from session):
|
||||||
|
|
||||||
|
The old (commented-out) `get_app_dir()` override gated
|
||||||
|
on `'pytest' in sys.modules` which can NEVER work in
|
||||||
|
spawned subactors (fresh procs, no pytest import); as
|
||||||
|
a result test `paperboi`/daemon actors were writing
|
||||||
|
into the user's REAL `~/.config/piker/accounting/`
|
||||||
|
files. Evidence: `account.kraken.paper.toml` gained a
|
||||||
|
single 0.001 xbtusdt clear stamped 2026-06-09T18:47Z
|
||||||
|
(a test fill), and `trades_kraken_paper.toml` (252B)
|
||||||
|
contained only that fill. The resolution must be
|
||||||
|
checked lazily at config-path access time (NOT import
|
||||||
|
time) since sub-actors only receive runtime-vars once
|
||||||
|
their `tractor` runtime has fully booted.
|
||||||
|
|
@ -0,0 +1,53 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:10:22Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171022Z_4485f2b9_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`). Proximate
|
||||||
|
driver: the approved plan's per-stage test gates could
|
||||||
|
not run AT ALL — the branch base was broken vs.
|
||||||
|
`tractor` git `main` (`AttributeError: 'tuple' object
|
||||||
|
has no attribute 'unwrap'` at `pikerd` boot, stale
|
||||||
|
`arbiter_sockaddr`/`.type`/`.size` API refs). The agent
|
||||||
|
fixed forward autonomously to (re)establish the gate
|
||||||
|
baseline, continuing the `repair_tests` branch lineage
|
||||||
|
(verified already merged in ancestry via `git cherry`
|
||||||
|
during a user-requested branch-overlap survey).
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Port the service layer + test suites to current
|
||||||
|
`tractor` APIs: addr-type normalization in
|
||||||
|
`open_pikerd()`, `query_actor()`/`wait_for_actor()`
|
||||||
|
kwarg renames, modern self-cancel absorption semantics
|
||||||
|
in the cancel-method test harness, exc/position attr
|
||||||
|
renames, a paper-EMS startup-budget bump and a syntax
|
||||||
|
fix in `.deribit.api`.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/service/_actor_runtime.py` — `wrap_address()`
|
||||||
|
normalize before `.unwrap()` in `open_pikerd()`
|
||||||
|
- `piker/service/_registry.py` — `check_for_service()`
|
||||||
|
-> `query_actor(regaddr=)` + 2-tuple yield +
|
||||||
|
`open_registry(addrs=)`
|
||||||
|
- `piker/brokers/deribit/api.py` — missing comma in
|
||||||
|
`tractor.trionics` import tuple
|
||||||
|
- `tests/test_services.py` — `registry_addr=` kwarg,
|
||||||
|
raddr unwraps, cancel-semantics harness rewrite,
|
||||||
|
`fail_after` 9 -> 19s
|
||||||
|
- `tests/test_ems.py` — `.boxed_type`, `pp.cumsize`
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,56 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:10:22Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/service/_actor_runtime.py`
|
||||||
|
|
||||||
|
Generated: normalize each registry addr via
|
||||||
|
`tractor.discovery._addr.wrap_address()` before
|
||||||
|
`.unwrap()`-ing for the `accept_addrs` bind check —
|
||||||
|
entries may be raw `tuple`s when passed in from (test)
|
||||||
|
client code. Import-path precedent taken from
|
||||||
|
`piker/cli/__init__.py:336`.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/service/_registry.py`
|
||||||
|
|
||||||
|
Generated: `check_for_service()` ported to
|
||||||
|
`tractor.query_actor(name, regaddr=...)` (kwarg was
|
||||||
|
`arbiter_sockaddr=`), unpacking the new
|
||||||
|
`(sockaddr, portal)` yield, and passing the
|
||||||
|
now-required `open_registry(addrs=Registry.addrs)`.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- tests/test_services.py`
|
||||||
|
> `git log -1 -p --follow -- tests/test_ems.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/deribit/api.py`
|
||||||
|
|
||||||
|
Key diagnostic reasoning (verbatim from session):
|
||||||
|
|
||||||
|
- the "DID NOT RAISE ContextCancelled" failure: in this
|
||||||
|
test the client actor IS pikerd (in-proc), and
|
||||||
|
current `tractor` main absorbs a `ContextCancelled`
|
||||||
|
whose canceller is your own actor — self-requested
|
||||||
|
cancels now exit cleanly instead of raising; the
|
||||||
|
'sigint' variant propagates a bare collapsed
|
||||||
|
`KeyboardInterrupt` rather than a
|
||||||
|
`BaseExceptionGroup`.
|
||||||
|
- the hard-coded `trio.fail_after(9)` startup budget is
|
||||||
|
marginal — full stack boot (pikerd -> emsd ->
|
||||||
|
brokerd.kraken -> paperboi + live kraken symbology
|
||||||
|
fetch) occasionally exceeds 9s -> bumped to 19s.
|
||||||
|
- `RemoteActorError.type` -> `.boxed_type`;
|
||||||
|
`Position.size` -> `.cumsize` (the paper engine
|
||||||
|
populates `BrokerdPosition.size` from `pp.cumsize`).
|
||||||
|
- overlap survey (user-requested): all of the
|
||||||
|
`repair_tests` branch commits are already in this
|
||||||
|
stack's ancestry; this commit finishes that branch's
|
||||||
|
port mission (its f4c4f1e2 fixed `conftest.py`'s
|
||||||
|
`arbiter_sockaddr` usage; this fixes the remaining
|
||||||
|
`test_services.py` + `check_for_service()` sites).
|
||||||
|
|
@ -0,0 +1,54 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:11:05Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171105Z_bc6e18d7_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`); this is the
|
||||||
|
approved plan's "stage 0" prep. Plan-shaping user
|
||||||
|
decisions captured via in-session Q&A:
|
||||||
|
|
||||||
|
- `datad.<broker>` topology: sibling of
|
||||||
|
`brokerd.<broker>` under `pikerd` (vs. child
|
||||||
|
subactor).
|
||||||
|
- migration: hard cutover, staged by layer (no
|
||||||
|
dual-mode runtime flag).
|
||||||
|
- post-split `brokerd` scope: trading-only +
|
||||||
|
EMS-lazy-spawned (charts/CLI never touch it).
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Declare per-daemon-kind backend submod groups
|
||||||
|
(`_datad_mods`/`_brokerd_mods`) in the split-style
|
||||||
|
backends, keyed to the pre-existing
|
||||||
|
`piker.data.validate._eps` contract; add a
|
||||||
|
`validate.get_eps()` introspection helper; ws-import
|
||||||
|
hygiene in `.kraken.broker`. Zero behavior change
|
||||||
|
(`__enable_modules__` unions unchanged).
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/data/validate.py` — add `get_eps()`
|
||||||
|
- `piker/brokers/kraken/__init__.py` — mod groups
|
||||||
|
- `piker/brokers/binance/__init__.py` — mod groups
|
||||||
|
(note: no `symbols.py`; search eps live in `.feed`)
|
||||||
|
- `piker/brokers/deribit/__init__.py` — datad-only
|
||||||
|
groups (no `broker.py` yet)
|
||||||
|
- `piker/brokers/ib/__init__.py` — add `'api'` to the
|
||||||
|
pre-existing `_datad_mods`
|
||||||
|
- `piker/brokers/kraken/broker.py` — import
|
||||||
|
`NoBsWs`/`open_autorecon_ws` from
|
||||||
|
`piker.data._web_bs` directly
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,46 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:11:05Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/data/validate.py`
|
||||||
|
|
||||||
|
Generated: `get_eps(mod, kind) -> dict[str, Callable]`
|
||||||
|
returning the daemon-kind's ep funcs defined by a
|
||||||
|
backend mod, keyed by ep name, missing eps excluded;
|
||||||
|
sourced from the existing `_eps` grouping table
|
||||||
|
(`'middleware' | 'datad' | 'brokerd'`).
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/kraken/__init__.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/binance/__init__.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/deribit/__init__.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/ib/__init__.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/kraken/broker.py`
|
||||||
|
|
||||||
|
Key exploration findings driving the design (from the
|
||||||
|
planning phase, 3 parallel explore agents + 1 architect
|
||||||
|
agent):
|
||||||
|
|
||||||
|
- the codebase already anticipated this split:
|
||||||
|
`validate._eps` groups eps into 'datad' vs 'brokerd'
|
||||||
|
kinds and `ib/__init__.py` already declared
|
||||||
|
`_brokerd_mods`/`_datad_mods`; `brokers/_daemon.py`
|
||||||
|
carried the literal TODO "rename the daemon to datad
|
||||||
|
prolly once we split up broker vs. data tasks into
|
||||||
|
separate actors?".
|
||||||
|
- the feared kraken feed<->broker "shared ws state"
|
||||||
|
hazard is mild: `NoBsWs`/`open_autorecon_ws`
|
||||||
|
originate in `piker.data._web_bs` (re-exported via
|
||||||
|
`.kraken.feed`); only `stream_messages` is feed-local
|
||||||
|
and it's a pure parser — in-process imports only,
|
||||||
|
each actor opens its own ws conn.
|
||||||
|
- `enable_modules` gates RPC entry, NOT python imports,
|
||||||
|
so backend trading mods may keep importing feed-side
|
||||||
|
helpers in-process post-split.
|
||||||
|
|
@ -0,0 +1,47 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:11:42Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171142Z_119d2c04_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`); this is the
|
||||||
|
approved plan's "stage 1": introduce the `datad` daemon
|
||||||
|
machinery additively (nothing routes through it yet).
|
||||||
|
User-decided constraint applied: `datad.<broker>` is a
|
||||||
|
SIBLING of `brokerd.<broker>` under `pikerd`, spawned
|
||||||
|
via the existing `Services` + `maybe_spawn_daemon()`
|
||||||
|
machinery.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
New `piker/data/_daemon.py` hosting the
|
||||||
|
`datad.<broker>` feed-only daemon-actor: lifetime
|
||||||
|
fixture owning the actor-global `_FeedsBus`, init/spawn
|
||||||
|
fns mirroring `.brokers._daemon` conventions and the
|
||||||
|
`samplerd` sub-daemon precedent, plus root-mod
|
||||||
|
registration, `piker.service` re-exports and a spawn
|
||||||
|
test.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/data/_daemon.py` — NEW:
|
||||||
|
`_setup_persistent_datad()`, `datad_init()`,
|
||||||
|
`spawn_datad()`, `maybe_spawn_datad()`,
|
||||||
|
`_datad_service_mods`
|
||||||
|
- `piker/service/_actor_runtime.py` — add
|
||||||
|
`piker.data._daemon` to `_root_modules`
|
||||||
|
- `piker/service/__init__.py` — re-export spawn eps
|
||||||
|
- `tests/test_services.py` — add `test_datad_spawn`
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,60 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:11:42Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/data/_daemon.py`
|
||||||
|
|
||||||
|
Generated symbols + key design decisions:
|
||||||
|
|
||||||
|
- `_datad_service_mods: list[str]` — datad-always
|
||||||
|
enabled mods, the data-side successor to the old
|
||||||
|
`piker.brokers._daemon._data_mods` set; kept minimal
|
||||||
|
per the caps-sec model.
|
||||||
|
- `_setup_persistent_datad()` — `@tractor.context`
|
||||||
|
lifetime fixture: console-log setup then allocates
|
||||||
|
the actor-global feed bus via
|
||||||
|
`feed.get_feed_bus(brokername, service_nursery)`
|
||||||
|
exactly as the old brokerd fixture did, pinned open
|
||||||
|
with `ctx.started()` + `sleep_forever()`.
|
||||||
|
- `datad_init()` — actor name `f'datad.{brokername}'`;
|
||||||
|
copies backend `_spawn_kwargs` (CRITICAL for `ib`'s
|
||||||
|
`infect_asyncio=True`); builds `enable_modules` from
|
||||||
|
`getattr(brokermod, '_datad_mods',
|
||||||
|
getattr(brokermod, '__enable_modules__', []))` as
|
||||||
|
the flat-backend fallback.
|
||||||
|
- `spawn_datad()` — `Services.actor_n.start_actor()` +
|
||||||
|
`Services.start_service_task()` exactly mirroring
|
||||||
|
`spawn_brokerd()`; dedup-composes enable mods via
|
||||||
|
`list(dict.fromkeys(...))`.
|
||||||
|
- `maybe_spawn_datad()` — wraps `maybe_spawn_daemon(
|
||||||
|
service_name=f'datad.{brokername}', ...)`.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/service/_actor_runtime.py`
|
||||||
|
> `git log -1 -p --follow -- piker/service/__init__.py`
|
||||||
|
> `git log -1 -p --follow -- tests/test_services.py`
|
||||||
|
|
||||||
|
Design rationale (verbatim from session):
|
||||||
|
|
||||||
|
- `_root_modules` must gain `piker.data._daemon` so
|
||||||
|
`pikerd_portal.run(spawn_datad, ...)` resolves in
|
||||||
|
the root.
|
||||||
|
- the `Services`-based impl style deliberately mirrors
|
||||||
|
`spawn_brokerd()` so the eventual `tractor.hilevel`
|
||||||
|
`ServiceMngr` port (see the `service_mng_to_tractor`
|
||||||
|
branch's d8c21d44 prep, surfaced by the
|
||||||
|
user-requested branch-overlap survey) lands
|
||||||
|
symmetrically on both spawn fns.
|
||||||
|
- mod placement (`piker/data/_daemon.py` vs.
|
||||||
|
generalizing `piker.brokers._daemon`) follows the
|
||||||
|
per-subsystem daemon-mod convention
|
||||||
|
(`.clearing._ems`, `.data._sampling`) and resolves
|
||||||
|
the existing TODO at `brokers/_daemon.py:49` ("move
|
||||||
|
this def to the `.data` subpkg").
|
||||||
|
|
@ -0,0 +1,47 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:12:26Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171226Z_64181219_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`); this is the
|
||||||
|
approved plan's "stage 2": decouple the clearing layer
|
||||||
|
from `feed.portals` BEFORE the feed cutover so live
|
||||||
|
trading works at every stage boundary. User-decided
|
||||||
|
constraint applied: post-split `brokerd` is
|
||||||
|
trading-only and spawned LAZILY only by `emsd`'s
|
||||||
|
`open_brokerd_dialog()` path.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Kill the single coupling forcing feed + trading eps
|
||||||
|
into one actor (`Router.open_trade_relays()` pulling
|
||||||
|
its trades portal from `feed.portals[brokermod]`):
|
||||||
|
`open_brokerd_dialog()` now (maybe) spawns/finds
|
||||||
|
`brokerd.<broker>` itself and ONLY when a live
|
||||||
|
trades-ep will actually open; paper mode never touches
|
||||||
|
it. Pre-cutover this is a pure refactor (registry
|
||||||
|
lookup finds the same feed-spawned daemon).
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/clearing/_ems.py` — `open_brokerd_dialog()`
|
||||||
|
re-sig + inner `acquire_live_portal()`;
|
||||||
|
`Router.maybe_open_brokerd_dialog()` drops `portal`
|
||||||
|
param; `open_trade_relays()` drops the
|
||||||
|
`feed.portals` lookup
|
||||||
|
- `piker/accounting/cli.py` — keyword-form `portal=`
|
||||||
|
override kept for the `piker ledger` ad-hoc actor
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:12:26Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/clearing/_ems.py`
|
||||||
|
|
||||||
|
Generated: new `open_brokerd_dialog()` signature
|
||||||
|
`(brokermod, exec_mode, fqme=None, portal=None,
|
||||||
|
loglevel=None)` with an inner `@acm
|
||||||
|
acquire_live_portal()` that yields the caller-provided
|
||||||
|
`portal` override (the `piker ledger` path) else
|
||||||
|
`maybe_spawn_brokerd(brokermod.name)` — designated THE
|
||||||
|
one place a live, credentialed `brokerd.<broker>` gets
|
||||||
|
booted post-split. The eager
|
||||||
|
`portal.open_context(trades_endpoint, ...)`
|
||||||
|
construction moved inside that block so the paper
|
||||||
|
short-circuit never acquires a live portal.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/accounting/cli.py`
|
||||||
|
|
||||||
|
Key analysis (verbatim from session):
|
||||||
|
|
||||||
|
- `feed.portals` had exactly ONE trading consumer:
|
||||||
|
`piker/clearing/_ems.py:671` (`portal =
|
||||||
|
feed.portals[brokermod]`) — the single coupling
|
||||||
|
forcing feed + trading into one actor.
|
||||||
|
- `piker ledger` (`accounting/cli.py:100-157`) is a
|
||||||
|
hidden consumer: it calls `broker_init()` directly,
|
||||||
|
spawns its own ad-hoc actor and passes the portal
|
||||||
|
positionally — the new signature must keep an
|
||||||
|
explicit `portal:` override param for this path.
|
||||||
|
- stage sequencing: clearing decouple lands BEFORE the
|
||||||
|
feed cutover, otherwise `feed.portals` would hand
|
||||||
|
emsd a `datad` portal and live `open_trade_dialog`
|
||||||
|
RPC would fail.
|
||||||
|
|
@ -0,0 +1,48 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:12:59Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171259Z_59d5d9a6_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`); this is the
|
||||||
|
approved plan's "stage 3": the feed-layer hard cutover
|
||||||
|
(user-decided: no dual-mode runtime flag, each stage
|
||||||
|
lands fully working).
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
The topology flip: all data-feed consumers route to
|
||||||
|
`datad.<broker>`; the `brokerd` lifetime fixture slims
|
||||||
|
to trading-only; UI/CLI data-flavoured spawn sites
|
||||||
|
repointed; test expectations inverted incl. an explicit
|
||||||
|
negative assert that paper sessions spawn NO
|
||||||
|
`brokerd.<broker>`.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/data/feed.py` — `open_feed()` ->
|
||||||
|
`maybe_spawn_datad()`; `open_feed_bus()` actor-name
|
||||||
|
assert -> `'datad'`; `install_brokerd_search` ->
|
||||||
|
`install_datad_search`; comment sweep
|
||||||
|
- `piker/brokers/_daemon.py` —
|
||||||
|
`_setup_persistent_brokerd()` slimmed (feed-bus
|
||||||
|
alloc moved to the datad fixture)
|
||||||
|
- `piker/ui/_app.py`, `piker/ui/cli.py`,
|
||||||
|
`piker/brokers/core.py`, `piker/brokers/cli.py`,
|
||||||
|
`piker/ui/kivy/option_chain.py` — spawn-site
|
||||||
|
repoints
|
||||||
|
- `tests/test_services.py` — datad assertions + the
|
||||||
|
no-`brokerd.kraken` negative check
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,54 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:12:59Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/data/feed.py`
|
||||||
|
|
||||||
|
Generated: `open_feed()` spawn cutover with the
|
||||||
|
`maybe_spawn_datad` import done "relative-direct" from
|
||||||
|
`._daemon` — NOT via `piker.service` — to dodge a
|
||||||
|
partial-init cycle: when `piker.data.feed` loads as
|
||||||
|
part of `piker.service.__init__` executing its own
|
||||||
|
`piker.data._daemon` import, the service pkg is
|
||||||
|
mid-init and its `maybe_spawn_datad` binding does not
|
||||||
|
exist yet. The `open_feed_bus()` local-state sanity
|
||||||
|
assert flips `'brokerd' in servicename` ->
|
||||||
|
`'datad' in servicename` (the only actor-name assert
|
||||||
|
in the tree, verified by grep).
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/_daemon.py`
|
||||||
|
|
||||||
|
Generated: `_setup_persistent_brokerd()` slimmed to
|
||||||
|
console-log setup + pinned-open ctx; drops the bus
|
||||||
|
alloc, the `assert not feed._bus`, the service nursery
|
||||||
|
(backend `open_trade_dialog()` ctxs own their task
|
||||||
|
trees) and the `eg.ExceptionGroup` handler. The
|
||||||
|
`piker ledger` ad-hoc actor enters this same slimmed
|
||||||
|
fixture — exactly what it needs.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/ui/_app.py`
|
||||||
|
> `git log -1 -p --follow -- piker/ui/cli.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/core.py`
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/cli.py`
|
||||||
|
> `git log -1 -p --follow -- piker/ui/kivy/option_chain.py`
|
||||||
|
> `git log -1 -p --follow -- tests/test_services.py`
|
||||||
|
|
||||||
|
Verification (from session): per-suite gates green at
|
||||||
|
this commit (services 5-passed incl. the new negative
|
||||||
|
assert, feeds 3-passed); a headless live smoke
|
||||||
|
(`maybe_open_pikerd` + `open_feed(['xbtusdt.kraken'])`
|
||||||
|
on an alt registry port) confirmed quotes flowing via
|
||||||
|
`datad.kraken` + `samplerd` with `check_for_service(
|
||||||
|
'brokerd.kraken') is None`. Known pre-existing flake
|
||||||
|
documented: `test_multi_fill_positions`' second
|
||||||
|
in-proc runtime boot wedges ~50% (zombie subactor w/
|
||||||
|
unread parent-IPC bytes); reproduced with the split
|
||||||
|
fully reverted so NOT a regression of this work.
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||||
|
timestamp: 2026-06-10T17:13:44Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260610T171344Z_eee19de0_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Same session-initiating `brokerd`-split instruction (see
|
||||||
|
`20260610T170859Z_75cefe10_prompt_io.md`); this is the
|
||||||
|
approved plan's final "stage 4": caps-sec slimming of
|
||||||
|
the (live, credentialed) trading actor + the `ib`
|
||||||
|
dual-daemon `client_id` collision mitigation flagged in
|
||||||
|
the plan's risk register.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
`brokerd` loses ALL `piker.data.*` (feed) RPC mods;
|
||||||
|
spawn fails fast for datad-only backends with a "use
|
||||||
|
paper-mode" error; `ib`'s default api-gw `client_id`
|
||||||
|
gets a per-daemon-kind offset so `datad.ib` +
|
||||||
|
`brokerd.ib` don't collide on connect.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `piker/brokers/_daemon.py` — `_data_mods` -> minimal
|
||||||
|
`_brokerd_service_mods`; `broker_init()` reads
|
||||||
|
`_brokerd_mods` (fallback `__enable_modules__`);
|
||||||
|
`spawn_brokerd()` fail-fast via `validate.get_eps()`
|
||||||
|
- `piker/brokers/ib/api.py` — role-based `client_id`
|
||||||
|
offset in `load_aio_clients()`
|
||||||
|
- `piker/cli/__init__.py` — resolved "expose datad"
|
||||||
|
TODO
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None — committed as generated.
|
||||||
|
|
@ -0,0 +1,41 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5[1m]
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-06-10T17:13:44Z
|
||||||
|
git_ref: datad_service
|
||||||
|
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
|
||||||
|
---
|
||||||
|
|
||||||
|
NOTE: diff-ref mode entry (code committed in the same
|
||||||
|
commit as this log); backfilled from the live dev
|
||||||
|
session transcript per the `/prompt-io` skill rules.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/_daemon.py`
|
||||||
|
|
||||||
|
Generated: the fail-fast originally landed in
|
||||||
|
`broker_init()` but was relocated to `spawn_brokerd()`
|
||||||
|
mid-implementation after realizing `piker ledger`
|
||||||
|
calls `broker_init()` directly even for paper accounts
|
||||||
|
on datad-only backends (would have crashed the cli);
|
||||||
|
the service-spawn path is the correct enforcement
|
||||||
|
seam. Error text:
|
||||||
|
|
||||||
|
Backend 'kucoin' offers NO `brokerd` (live
|
||||||
|
order-control) eps!? It is likely a datad-only
|
||||||
|
provider, use paper-mode for clearing instead.
|
||||||
|
|
||||||
|
(verified live via a `trio.run()` unit check.)
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/brokers/ib/api.py`
|
||||||
|
|
||||||
|
Generated: in `load_aio_clients()`, when `client_id`
|
||||||
|
is the 6116 default: `datad`-named actors offset +16
|
||||||
|
(disjoint from `brokerd`'s linear `client_id + i`
|
||||||
|
retry range), other non-`brokerd` (ad-hoc test/cli)
|
||||||
|
actors +32. Rationale from the plan's risk register:
|
||||||
|
post-split BOTH per-broker daemons connect to the same
|
||||||
|
TWS/gw endpoint; a shared default id collides and
|
||||||
|
burns up to `connect_timeout * retries` (90s) in
|
||||||
|
retry cycles.
|
||||||
|
|
||||||
|
> `git log -1 -p --follow -- piker/cli/__init__.py`
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
# AI Prompt I/O Log — claude
|
||||||
|
|
||||||
|
This directory tracks prompt inputs and model
|
||||||
|
outputs for AI-assisted development using
|
||||||
|
`claude` (claude-code CLI).
|
||||||
|
|
||||||
|
## Policy
|
||||||
|
|
||||||
|
Prompt logging follows the
|
||||||
|
[NLNet generative AI policy][nlnet-ai].
|
||||||
|
All substantive AI contributions are logged
|
||||||
|
with:
|
||||||
|
- Model name and version
|
||||||
|
- Timestamps
|
||||||
|
- The prompts that produced the output
|
||||||
|
- Unedited model output (`.raw.md` files)
|
||||||
|
|
||||||
|
[nlnet-ai]: https://nlnet.nl/foundation/policies/generativeAI/
|
||||||
|
|
||||||
|
## Usage
|
||||||
|
|
||||||
|
Entries are created by the `/prompt-io` skill
|
||||||
|
or automatically via `/commit-msg` integration.
|
||||||
|
|
||||||
|
Each commit carrying AI-generated changes links
|
||||||
|
to its provenance entry via a `Prompt-IO:`
|
||||||
|
commit-msg trailer; entries use "diff-ref mode"
|
||||||
|
(pointers into `git log -p` instead of verbatim
|
||||||
|
code copies) to avoid duplicating committed
|
||||||
|
code.
|
||||||
|
|
||||||
|
Human contributors remain accountable for all
|
||||||
|
code decisions. AI-generated content is never
|
||||||
|
presented as human-authored work.
|
||||||
|
|
@ -324,10 +324,13 @@ def load_ledger(
|
||||||
ldir: Path = (
|
ldir: Path = (
|
||||||
dirpath
|
dirpath
|
||||||
or
|
or
|
||||||
config._config_dir / 'accounting' / 'ledgers'
|
config.get_conf_dir() / 'accounting' / 'ledgers'
|
||||||
)
|
)
|
||||||
if not ldir.is_dir():
|
if not ldir.is_dir():
|
||||||
ldir.mkdir()
|
ldir.mkdir(
|
||||||
|
parents=True,
|
||||||
|
exist_ok=True,
|
||||||
|
)
|
||||||
|
|
||||||
fname = f'trades_{brokername}_{acctid}.toml'
|
fname = f'trades_{brokername}_{acctid}.toml'
|
||||||
fpath: Path = ldir / fname
|
fpath: Path = ldir / fname
|
||||||
|
|
|
||||||
|
|
@ -785,9 +785,16 @@ def load_account(
|
||||||
legacy_fn: str = f'pps.{brokername}.{acctid}.toml'
|
legacy_fn: str = f'pps.{brokername}.{acctid}.toml'
|
||||||
fn: str = f'account.{brokername}.{acctid}.toml'
|
fn: str = f'account.{brokername}.{acctid}.toml'
|
||||||
|
|
||||||
dirpath: Path = dirpath or (config._config_dir / 'accounting')
|
dirpath: Path = (
|
||||||
|
dirpath
|
||||||
|
or
|
||||||
|
(config.get_conf_dir() / 'accounting')
|
||||||
|
)
|
||||||
if not dirpath.is_dir():
|
if not dirpath.is_dir():
|
||||||
dirpath.mkdir()
|
dirpath.mkdir(
|
||||||
|
parents=True,
|
||||||
|
exist_ok=True,
|
||||||
|
)
|
||||||
|
|
||||||
conf, path = config.load(
|
conf, path = config.load(
|
||||||
path=dirpath / fn,
|
path=dirpath / fn,
|
||||||
|
|
|
||||||
|
|
@ -143,12 +143,15 @@ def sync(
|
||||||
# (what the EMS normally does internall) B)
|
# (what the EMS normally does internall) B)
|
||||||
open_brokerd_dialog(
|
open_brokerd_dialog(
|
||||||
brokermod,
|
brokermod,
|
||||||
portal,
|
|
||||||
exec_mode=(
|
exec_mode=(
|
||||||
'paper'
|
'paper'
|
||||||
if account == 'paper'
|
if account == 'paper'
|
||||||
else 'live'
|
else 'live'
|
||||||
),
|
),
|
||||||
|
# use our own ad-hoc-spawned actor,
|
||||||
|
# do NOT (spawn and) use the
|
||||||
|
# `brokerd.<broker>` service daemon!
|
||||||
|
portal=portal,
|
||||||
loglevel=loglevel,
|
loglevel=loglevel,
|
||||||
) as (
|
) as (
|
||||||
brokerd_stream,
|
brokerd_stream,
|
||||||
|
|
|
||||||
|
|
@ -25,10 +25,8 @@ from contextlib import (
|
||||||
)
|
)
|
||||||
from types import ModuleType
|
from types import ModuleType
|
||||||
from typing import (
|
from typing import (
|
||||||
TYPE_CHECKING,
|
|
||||||
AsyncContextManager,
|
AsyncContextManager,
|
||||||
)
|
)
|
||||||
import exceptiongroup as eg
|
|
||||||
|
|
||||||
import tractor
|
import tractor
|
||||||
import trio
|
import trio
|
||||||
|
|
@ -40,27 +38,20 @@ from piker.log import (
|
||||||
from . import _util
|
from . import _util
|
||||||
from . import get_brokermod
|
from . import get_brokermod
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from ..data import _FeedsBus
|
|
||||||
|
|
||||||
log = get_logger(name=__name__)
|
log = get_logger(name=__name__)
|
||||||
|
|
||||||
# `brokerd` enabled modules
|
# `brokerd`-actor-always-enabled mods.
|
||||||
# TODO: move this def to the `.data` subpkg..
|
|
||||||
# NOTE: keeping this list as small as possible is part of our caps-sec
|
# NOTE: keeping this list as small as possible is part of our caps-sec
|
||||||
# model and should be treated with utmost care!
|
# model and should be treated with utmost care! In particular NO
|
||||||
_data_mods: str = [
|
# `piker.data.*` feed mods should be enabled in this (live,
|
||||||
'piker.brokers.core',
|
# credentialed) trading actor; all data-feed serving is the
|
||||||
'piker.brokers.data',
|
# domain of the `datad.<broker>` sibling daemon, see
|
||||||
|
# `piker.data._daemon._datad_service_mods`.
|
||||||
|
_brokerd_service_mods: list[str] = [
|
||||||
'piker.brokers._daemon',
|
'piker.brokers._daemon',
|
||||||
'piker.data',
|
|
||||||
'piker.data.feed',
|
|
||||||
'piker.data._sampling'
|
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
# TODO: we should rename the daemon to datad prolly once we split up
|
|
||||||
# broker vs. data tasks into separate actors?
|
|
||||||
@tractor.context
|
@tractor.context
|
||||||
async def _setup_persistent_brokerd(
|
async def _setup_persistent_brokerd(
|
||||||
ctx: tractor.Context,
|
ctx: tractor.Context,
|
||||||
|
|
@ -69,9 +60,15 @@ async def _setup_persistent_brokerd(
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
'''
|
'''
|
||||||
Allocate a actor-wide service nursery in ``brokerd``
|
Trading-only daemon (lifetime) fixture: console logging
|
||||||
such that feeds can be run in the background persistently by
|
setup and a pinned-open context for service mgmt.
|
||||||
the broker backend as needed.
|
|
||||||
|
All data-feed-bus state now lives in the (data-feed-only)
|
||||||
|
`datad.<brokername>` sibling daemon, see
|
||||||
|
`piker.data._daemon._setup_persistent_datad()`; this
|
||||||
|
actor hosts only the backend's `open_trade_dialog()`
|
||||||
|
(live order-control) ep-task(s) which manage their own
|
||||||
|
task trees per `tractor.Context`.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# NOTE: we only need to setup logging once (and only) here
|
# NOTE: we only need to setup logging once (and only) here
|
||||||
|
|
@ -87,46 +84,12 @@ async def _setup_persistent_brokerd(
|
||||||
)
|
)
|
||||||
assert log.name == _util.subsys
|
assert log.name == _util.subsys
|
||||||
|
|
||||||
# further, set the log level on any broker broker specific
|
# unblock caller
|
||||||
# logger instance.
|
await ctx.started()
|
||||||
|
|
||||||
from piker.data import feed
|
# we pin this task to keep the daemon active until the
|
||||||
assert not feed._bus
|
# parent actor decides to tear it down
|
||||||
|
await trio.sleep_forever()
|
||||||
# allocate a nursery to the bus for spawning background
|
|
||||||
# tasks to service client IPC requests, normally
|
|
||||||
# `tractor.Context` connections to explicitly required
|
|
||||||
# `brokerd` endpoints such as:
|
|
||||||
# - `stream_quotes()`,
|
|
||||||
# - `manage_history()`,
|
|
||||||
# - `allocate_persistent_feed()`,
|
|
||||||
# - `open_symbol_search()`
|
|
||||||
# NOTE: see ep invocation details inside `.data.feed`.
|
|
||||||
try:
|
|
||||||
async with (
|
|
||||||
# tractor.trionics.collapse_eg(),
|
|
||||||
trio.open_nursery() as service_nursery
|
|
||||||
):
|
|
||||||
bus: _FeedsBus = feed.get_feed_bus(
|
|
||||||
brokername,
|
|
||||||
service_nursery,
|
|
||||||
)
|
|
||||||
assert bus is feed._bus
|
|
||||||
|
|
||||||
# unblock caller
|
|
||||||
await ctx.started()
|
|
||||||
|
|
||||||
# we pin this task to keep the feeds manager active until the
|
|
||||||
# parent actor decides to tear it down
|
|
||||||
await trio.sleep_forever()
|
|
||||||
|
|
||||||
except eg.ExceptionGroup:
|
|
||||||
# TODO: likely some underlying `brokerd` IPC connection
|
|
||||||
# broke so here we handle a respawn and re-connect attempt!
|
|
||||||
# This likely should pair with development of the OCO task
|
|
||||||
# nusery in dev over @ `tractor` B)
|
|
||||||
# https://github.com/goodboy/tractor/pull/363
|
|
||||||
raise
|
|
||||||
|
|
||||||
|
|
||||||
def broker_init(
|
def broker_init(
|
||||||
|
|
@ -147,8 +110,10 @@ def broker_init(
|
||||||
|
|
||||||
This includes:
|
This includes:
|
||||||
- load the appropriate <brokername>.py pkg module,
|
- load the appropriate <brokername>.py pkg module,
|
||||||
- reads any declared `__enable_modules__: listr[str]` which will be
|
- reads any declared `_brokerd_mods: list[str]` (falling
|
||||||
passed to `tractor.ActorNursery.start_actor(enabled_modules=<this>)`
|
back to the full `__enable_modules__` set for
|
||||||
|
not-yet-split backends) which will be passed to
|
||||||
|
`tractor.ActorNursery.start_actor(enable_modules=)`
|
||||||
at actor start time,
|
at actor start time,
|
||||||
- deliver a references to the daemon lifetime fixture, which
|
- deliver a references to the daemon lifetime fixture, which
|
||||||
for now is always the `_setup_persistent_brokerd()` context defined
|
for now is always the `_setup_persistent_brokerd()` context defined
|
||||||
|
|
@ -183,8 +148,14 @@ def broker_init(
|
||||||
]
|
]
|
||||||
for submodname in getattr(
|
for submodname in getattr(
|
||||||
brokermod,
|
brokermod,
|
||||||
'__enable_modules__',
|
'_brokerd_mods',
|
||||||
[],
|
# fallback for (flat, less mature) backends which
|
||||||
|
# don't yet declare a daemon-kind mod split.
|
||||||
|
getattr(
|
||||||
|
brokermod,
|
||||||
|
'__enable_modules__',
|
||||||
|
[],
|
||||||
|
),
|
||||||
):
|
):
|
||||||
subpath: str = f'{modpath}.{submodname}'
|
subpath: str = f'{modpath}.{submodname}'
|
||||||
enabled.append(subpath)
|
enabled.append(subpath)
|
||||||
|
|
@ -212,6 +183,22 @@ async def spawn_brokerd(
|
||||||
f'backend: {brokername!r}'
|
f'backend: {brokername!r}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# fail fast on (data-only) backends which don't offer
|
||||||
|
# ANY live order-control eps; the caller should instead
|
||||||
|
# be using paper-mode (and thus never spawning us)!
|
||||||
|
from ..data.validate import get_eps
|
||||||
|
brokerd_eps: dict = get_eps(
|
||||||
|
get_brokermod(brokername),
|
||||||
|
'brokerd',
|
||||||
|
)
|
||||||
|
if not brokerd_eps:
|
||||||
|
raise RuntimeError(
|
||||||
|
f'Backend {brokername!r} offers NO `brokerd` '
|
||||||
|
f'(live order-control) eps!?\n'
|
||||||
|
f'It is likely a datad-only provider, use '
|
||||||
|
f'paper-mode for clearing instead.\n'
|
||||||
|
)
|
||||||
|
|
||||||
(
|
(
|
||||||
brokermode,
|
brokermode,
|
||||||
tractor_kwargs,
|
tractor_kwargs,
|
||||||
|
|
@ -233,7 +220,11 @@ async def spawn_brokerd(
|
||||||
dname: str = tractor_kwargs.pop('name') # f'brokerd.{brokername}'
|
dname: str = tractor_kwargs.pop('name') # f'brokerd.{brokername}'
|
||||||
portal = await Services.actor_n.start_actor(
|
portal = await Services.actor_n.start_actor(
|
||||||
dname,
|
dname,
|
||||||
enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
|
enable_modules=list(dict.fromkeys(
|
||||||
|
_brokerd_service_mods
|
||||||
|
+
|
||||||
|
tractor_kwargs.pop('enable_modules')
|
||||||
|
)),
|
||||||
debug_mode=Services.debug_mode,
|
debug_mode=Services.debug_mode,
|
||||||
**tractor_kwargs
|
**tractor_kwargs
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -52,9 +52,25 @@ __all__ = [
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
# `brokerd` modules
|
# per-daemon-kind (sub)mod groups: declares which of our
|
||||||
__enable_modules__: list[str] = [
|
# submods host the eps run by each daemon-actor kind as
|
||||||
|
# defined by `piker.data.validate._eps`.
|
||||||
|
# NOTE: `get_mkt_info` and `open_symbol_search` both live
|
||||||
|
# in `.feed` for this backend (no `symbols.py`).
|
||||||
|
_brokerd_mods: list[str] = [
|
||||||
'api',
|
'api',
|
||||||
'feed',
|
|
||||||
'broker',
|
'broker',
|
||||||
]
|
]
|
||||||
|
|
||||||
|
_datad_mods: list[str] = [
|
||||||
|
'api',
|
||||||
|
'feed',
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
# tractor RPC enable arg
|
||||||
|
__enable_modules__: list[str] = list(dict.fromkeys(
|
||||||
|
_brokerd_mods
|
||||||
|
+
|
||||||
|
_datad_mods
|
||||||
|
))
|
||||||
|
|
|
||||||
|
|
@ -35,7 +35,7 @@ from piker.log import (
|
||||||
get_logger,
|
get_logger,
|
||||||
)
|
)
|
||||||
from ..service import (
|
from ..service import (
|
||||||
maybe_spawn_brokerd,
|
maybe_spawn_datad,
|
||||||
maybe_open_pikerd,
|
maybe_open_pikerd,
|
||||||
)
|
)
|
||||||
from ..brokers import (
|
from ..brokers import (
|
||||||
|
|
@ -187,7 +187,7 @@ def brokercheck(config, broker):
|
||||||
|
|
||||||
'''
|
'''
|
||||||
async def bcheck_main():
|
async def bcheck_main():
|
||||||
async with maybe_spawn_brokerd(broker) as portal:
|
async with maybe_spawn_datad(broker) as portal:
|
||||||
await portal.run(run_test, broker)
|
await portal.run(run_test, broker)
|
||||||
await portal.cancel_actor()
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
@ -317,7 +317,7 @@ def record(config, rate, name, dhost, filename):
|
||||||
return
|
return
|
||||||
|
|
||||||
async def main(tries):
|
async def main(tries):
|
||||||
async with maybe_spawn_brokerd(
|
async with maybe_spawn_datad(
|
||||||
tries=tries, loglevel=loglevel
|
tries=tries, loglevel=loglevel
|
||||||
) as portal:
|
) as portal:
|
||||||
# run app "main"
|
# run app "main"
|
||||||
|
|
|
||||||
|
|
@ -30,7 +30,7 @@ import trio
|
||||||
|
|
||||||
from piker.log import get_logger
|
from piker.log import get_logger
|
||||||
from . import get_brokermod
|
from . import get_brokermod
|
||||||
from ..service import maybe_spawn_brokerd
|
from ..service import maybe_spawn_datad
|
||||||
from . import open_cached_client
|
from . import open_cached_client
|
||||||
from ..accounting import MktPair
|
from ..accounting import MktPair
|
||||||
|
|
||||||
|
|
@ -172,7 +172,7 @@ async def symbol_search(
|
||||||
# await tractor.devx._debug.maybe_init_greenback()
|
# await tractor.devx._debug.maybe_init_greenback()
|
||||||
# tractor.pause_from_sync()
|
# tractor.pause_from_sync()
|
||||||
|
|
||||||
async with maybe_spawn_brokerd(
|
async with maybe_spawn_datad(
|
||||||
mod.name,
|
mod.name,
|
||||||
infect_asyncio=getattr(
|
infect_asyncio=getattr(
|
||||||
mod,
|
mod,
|
||||||
|
|
|
||||||
|
|
@ -47,13 +47,25 @@ __all__ = [
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
# tractor RPC enable arg
|
# per-daemon-kind (sub)mod groups: declares which of our
|
||||||
__enable_modules__: list[str] = [
|
# submods host the eps run by each daemon-actor kind as
|
||||||
|
# defined by `piker.data.validate._eps`.
|
||||||
|
# NOTE: datad-only backend (no `broker.py` yet)!
|
||||||
|
_brokerd_mods: list[str] = []
|
||||||
|
|
||||||
|
_datad_mods: list[str] = [
|
||||||
'api',
|
'api',
|
||||||
'feed',
|
'feed',
|
||||||
# 'broker',
|
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
# tractor RPC enable arg
|
||||||
|
__enable_modules__: list[str] = list(dict.fromkeys(
|
||||||
|
_brokerd_mods
|
||||||
|
+
|
||||||
|
_datad_mods
|
||||||
|
))
|
||||||
|
|
||||||
# passed to ``tractor.ActorNursery.start_actor()``
|
# passed to ``tractor.ActorNursery.start_actor()``
|
||||||
_spawn_kwargs = {
|
_spawn_kwargs = {
|
||||||
'infect_asyncio': True,
|
'infect_asyncio': True,
|
||||||
|
|
|
||||||
|
|
@ -37,7 +37,7 @@ from rapidfuzz import process as fuzzy
|
||||||
import numpy as np
|
import numpy as np
|
||||||
from tractor.trionics import (
|
from tractor.trionics import (
|
||||||
broadcast_receiver,
|
broadcast_receiver,
|
||||||
maybe_open_context
|
maybe_open_context,
|
||||||
collapse_eg,
|
collapse_eg,
|
||||||
)
|
)
|
||||||
from tractor import to_asyncio
|
from tractor import to_asyncio
|
||||||
|
|
|
||||||
|
|
@ -65,17 +65,18 @@ _brokerd_mods: list[str] = [
|
||||||
]
|
]
|
||||||
|
|
||||||
_datad_mods: list[str] = [
|
_datad_mods: list[str] = [
|
||||||
|
'api',
|
||||||
'feed',
|
'feed',
|
||||||
'symbols',
|
'symbols',
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
# tractor RPC enable arg
|
# tractor RPC enable arg
|
||||||
__enable_modules__: list[str] = (
|
__enable_modules__: list[str] = list(dict.fromkeys(
|
||||||
_brokerd_mods
|
_brokerd_mods
|
||||||
+
|
+
|
||||||
_datad_mods
|
_datad_mods
|
||||||
)
|
))
|
||||||
|
|
||||||
# passed to ``tractor.ActorNursery.start_actor()``
|
# passed to ``tractor.ActorNursery.start_actor()``
|
||||||
_spawn_kwargs = {
|
_spawn_kwargs = {
|
||||||
|
|
|
||||||
|
|
@ -1340,6 +1340,21 @@ async def load_aio_clients(
|
||||||
ib = None
|
ib = None
|
||||||
client = None
|
client = None
|
||||||
|
|
||||||
|
# XXX: post (brokerd vs. datad)-split BOTH per-broker
|
||||||
|
# daemons connect to the same API gw/tws endpoint(s); to
|
||||||
|
# avoid `clientId` collisions (and the long conn-timeout
|
||||||
|
# retry cycle they cause) we offset the data-daemon's
|
||||||
|
# default id-range to be disjoint from `brokerd.ib`'s
|
||||||
|
# (which also retries with `client_id + i` increments).
|
||||||
|
if client_id == 6116: # the default from above
|
||||||
|
aname: str = tractor.current_actor().name
|
||||||
|
if 'datad' in aname:
|
||||||
|
client_id += 16
|
||||||
|
# ad-hoc (test/cli) actors get their own range to
|
||||||
|
# avoid clashing with any live daemon-tree's conns.
|
||||||
|
elif 'brokerd' not in aname:
|
||||||
|
client_id += 32
|
||||||
|
|
||||||
# attempt to get connection info from config; if no .toml entry
|
# attempt to get connection info from config; if no .toml entry
|
||||||
# exists, we try to load from a default localhost connection.
|
# exists, we try to load from a default localhost connection.
|
||||||
localhost = '127.0.0.1'
|
localhost = '127.0.0.1'
|
||||||
|
|
|
||||||
|
|
@ -66,10 +66,24 @@ __all__ = [
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
# tractor RPC enable arg
|
# per-daemon-kind (sub)mod groups: declares which of our
|
||||||
__enable_modules__: list[str] = [
|
# submods host the eps run by each daemon-actor kind as
|
||||||
|
# defined by `piker.data.validate._eps`.
|
||||||
|
_brokerd_mods: list[str] = [
|
||||||
'api',
|
'api',
|
||||||
'broker',
|
'broker',
|
||||||
|
]
|
||||||
|
|
||||||
|
_datad_mods: list[str] = [
|
||||||
|
'api',
|
||||||
'feed',
|
'feed',
|
||||||
'symbols',
|
'symbols',
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
# tractor RPC enable arg
|
||||||
|
__enable_modules__: list[str] = list(dict.fromkeys(
|
||||||
|
_brokerd_mods
|
||||||
|
+
|
||||||
|
_datad_mods
|
||||||
|
))
|
||||||
|
|
|
||||||
|
|
@ -73,10 +73,12 @@ from piker.log import (
|
||||||
get_logger,
|
get_logger,
|
||||||
)
|
)
|
||||||
from piker.data import open_symcache
|
from piker.data import open_symcache
|
||||||
from . import api
|
from piker.data._web_bs import (
|
||||||
from .feed import (
|
|
||||||
open_autorecon_ws,
|
open_autorecon_ws,
|
||||||
NoBsWs,
|
NoBsWs,
|
||||||
|
)
|
||||||
|
from . import api
|
||||||
|
from .feed import (
|
||||||
stream_messages,
|
stream_messages,
|
||||||
)
|
)
|
||||||
from .ledger import (
|
from .ledger import (
|
||||||
|
|
|
||||||
|
|
@ -335,9 +335,14 @@ class TradesRelay(Struct):
|
||||||
@acm
|
@acm
|
||||||
async def open_brokerd_dialog(
|
async def open_brokerd_dialog(
|
||||||
brokermod: ModuleType,
|
brokermod: ModuleType,
|
||||||
portal: tractor.Portal,
|
|
||||||
exec_mode: str,
|
exec_mode: str,
|
||||||
fqme: str|None = None,
|
fqme: str|None = None,
|
||||||
|
|
||||||
|
# XXX: explicit (already spawned) trading-actor override,
|
||||||
|
# currently only used by the `piker ledger` cli which
|
||||||
|
# boots its own ad-hoc `brokerd`-like actor; normally we
|
||||||
|
# (lazily) spawn/find the `brokerd.<broker>` daemon here.
|
||||||
|
portal: tractor.Portal|None = None,
|
||||||
loglevel: str|None = None,
|
loglevel: str|None = None,
|
||||||
|
|
||||||
) -> tuple[
|
) -> tuple[
|
||||||
|
|
@ -351,6 +356,10 @@ async def open_brokerd_dialog(
|
||||||
paper engine instance depending on live trading support for the
|
paper engine instance depending on live trading support for the
|
||||||
broker backend, configuration, or client code usage.
|
broker backend, configuration, or client code usage.
|
||||||
|
|
||||||
|
NOTE: this is now the ONE place where a (live, credentialed)
|
||||||
|
`brokerd.<broker>` daemon-actor gets (lazily) booted; pure
|
||||||
|
data/paper sessions should never spawn one!
|
||||||
|
|
||||||
'''
|
'''
|
||||||
get_console_log(
|
get_console_log(
|
||||||
level=loglevel,
|
level=loglevel,
|
||||||
|
|
@ -416,16 +425,29 @@ async def open_brokerd_dialog(
|
||||||
)
|
)
|
||||||
exec_mode: str = 'paper'
|
exec_mode: str = 'paper'
|
||||||
|
|
||||||
if (
|
@acm
|
||||||
trades_endpoint is not None
|
async def acquire_live_portal():
|
||||||
or
|
'''
|
||||||
exec_mode != 'paper'
|
Deliver a portal to the (live, credentialed) trading
|
||||||
):
|
actor hosting the backend's `open_trade_dialog()` ep:
|
||||||
# open live brokerd trades endpoint
|
either the caller-provided override or the (maybe
|
||||||
open_trades_endpoint = portal.open_context(
|
lazily spawned) `brokerd.<broker>` service daemon.
|
||||||
trades_endpoint,
|
|
||||||
loglevel=loglevel,
|
'''
|
||||||
|
if portal is not None:
|
||||||
|
yield portal
|
||||||
|
return
|
||||||
|
|
||||||
|
# XXX: the ONE (normal) place a `brokerd.<broker>`
|
||||||
|
# daemon-actor gets booted in the runtime B)
|
||||||
|
from piker.brokers._daemon import (
|
||||||
|
maybe_spawn_brokerd,
|
||||||
)
|
)
|
||||||
|
async with maybe_spawn_brokerd(
|
||||||
|
brokermod.name,
|
||||||
|
loglevel=loglevel,
|
||||||
|
) as live_portal:
|
||||||
|
yield live_portal
|
||||||
|
|
||||||
@acm
|
@acm
|
||||||
async def maybe_open_paper_ep():
|
async def maybe_open_paper_ep():
|
||||||
|
|
@ -437,7 +459,14 @@ async def open_brokerd_dialog(
|
||||||
return
|
return
|
||||||
|
|
||||||
# open trades-dialog endpoint with backend broker
|
# open trades-dialog endpoint with backend broker
|
||||||
async with open_trades_endpoint as msg:
|
async with (
|
||||||
|
acquire_live_portal() as live_portal,
|
||||||
|
|
||||||
|
live_portal.open_context(
|
||||||
|
trades_endpoint,
|
||||||
|
loglevel=loglevel,
|
||||||
|
) as msg,
|
||||||
|
):
|
||||||
ctx, first = msg
|
ctx, first = msg
|
||||||
|
|
||||||
# runtime indication that the backend can't support live
|
# runtime indication that the backend can't support live
|
||||||
|
|
@ -581,7 +610,6 @@ class Router(Struct):
|
||||||
async def maybe_open_brokerd_dialog(
|
async def maybe_open_brokerd_dialog(
|
||||||
self,
|
self,
|
||||||
brokermod: ModuleType,
|
brokermod: ModuleType,
|
||||||
portal: tractor.Portal,
|
|
||||||
exec_mode: str,
|
exec_mode: str,
|
||||||
fqme: str,
|
fqme: str,
|
||||||
loglevel: str,
|
loglevel: str,
|
||||||
|
|
@ -606,7 +634,6 @@ class Router(Struct):
|
||||||
|
|
||||||
async with open_brokerd_dialog(
|
async with open_brokerd_dialog(
|
||||||
brokermod=brokermod,
|
brokermod=brokermod,
|
||||||
portal=portal,
|
|
||||||
exec_mode=exec_mode,
|
exec_mode=exec_mode,
|
||||||
fqme=fqme,
|
fqme=fqme,
|
||||||
loglevel=loglevel,
|
loglevel=loglevel,
|
||||||
|
|
@ -668,7 +695,6 @@ class Router(Struct):
|
||||||
brokername, _, _, _ = unpack_fqme(fqme)
|
brokername, _, _, _ = unpack_fqme(fqme)
|
||||||
brokermod = feed.mods[brokername]
|
brokermod = feed.mods[brokername]
|
||||||
broker = brokermod.name
|
broker = brokermod.name
|
||||||
portal = feed.portals[brokermod]
|
|
||||||
|
|
||||||
# XXX: this should be initial price quote from target provider
|
# XXX: this should be initial price quote from target provider
|
||||||
flume = feed.flumes[fqme]
|
flume = feed.flumes[fqme]
|
||||||
|
|
@ -682,7 +708,6 @@ class Router(Struct):
|
||||||
|
|
||||||
async with self.maybe_open_brokerd_dialog(
|
async with self.maybe_open_brokerd_dialog(
|
||||||
brokermod=brokermod,
|
brokermod=brokermod,
|
||||||
portal=portal,
|
|
||||||
exec_mode=exec_mode,
|
exec_mode=exec_mode,
|
||||||
fqme=fqme,
|
fqme=fqme,
|
||||||
loglevel=loglevel,
|
loglevel=loglevel,
|
||||||
|
|
|
||||||
|
|
@ -250,7 +250,6 @@ def cli(
|
||||||
|
|
||||||
# TODO: load endpoints from `conf::[network].pikerd`
|
# TODO: load endpoints from `conf::[network].pikerd`
|
||||||
# - pikerd vs. regd, separate registry daemon?
|
# - pikerd vs. regd, separate registry daemon?
|
||||||
# - expose datad vs. brokerd?
|
|
||||||
# - bind emsd with certain perms on public iface?
|
# - bind emsd with certain perms on public iface?
|
||||||
regaddrs: list[tuple[str, int]] = regaddr or [(
|
regaddrs: list[tuple[str, int]] = regaddr or [(
|
||||||
_default_registry_host,
|
_default_registry_host,
|
||||||
|
|
|
||||||
|
|
@ -188,6 +188,48 @@ def _override_config_dir(
|
||||||
_config_dir = path
|
_config_dir = path
|
||||||
|
|
||||||
|
|
||||||
|
def _maybe_use_test_dir() -> None:
|
||||||
|
'''
|
||||||
|
When running under the `pytest` harness, override
|
||||||
|
the config dir to the per-test-tmp dir "passed down"
|
||||||
|
the actor tree via `tractor`'s runtime-vars
|
||||||
|
inheritance mechanism.
|
||||||
|
|
||||||
|
See the `tractor_runtime_overrides` usage in our
|
||||||
|
`tests.conftest._open_test_pikerd()` as well as
|
||||||
|
`.service._actor_runtime.open_piker_runtime()` for
|
||||||
|
the root-actor's pre-loading of the var state.
|
||||||
|
|
||||||
|
NOTE: this must be checked lazily at config-path
|
||||||
|
access time (NOT import time) since sub-actors only
|
||||||
|
receive runtime-vars once their `tractor` runtime
|
||||||
|
has fully booted.
|
||||||
|
|
||||||
|
'''
|
||||||
|
global _config_dir
|
||||||
|
import tractor
|
||||||
|
actor = tractor.current_actor(
|
||||||
|
err_on_no_runtime=False,
|
||||||
|
)
|
||||||
|
if actor is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
rvs: dict = tractor.runtime._state._runtime_vars
|
||||||
|
pvars: dict|None = rvs.get('piker_vars')
|
||||||
|
if (
|
||||||
|
pvars
|
||||||
|
and
|
||||||
|
(testdir := pvars.get('piker_test_dir'))
|
||||||
|
):
|
||||||
|
testdirpath = Path(testdir)
|
||||||
|
assert testdirpath.exists(), (
|
||||||
|
f'piker test harness might be borked!?\n'
|
||||||
|
f'testdirpath: {testdirpath!r}\n'
|
||||||
|
)
|
||||||
|
if _config_dir != testdirpath:
|
||||||
|
_override_config_dir(testdirpath)
|
||||||
|
|
||||||
|
|
||||||
def _conf_fn_w_ext(
|
def _conf_fn_w_ext(
|
||||||
name: str,
|
name: str,
|
||||||
) -> str:
|
) -> str:
|
||||||
|
|
@ -201,6 +243,7 @@ def get_conf_dir() -> Path:
|
||||||
on the local filesystem.
|
on the local filesystem.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
|
_maybe_use_test_dir()
|
||||||
return _config_dir
|
return _config_dir
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -226,7 +269,7 @@ def get_conf_path(
|
||||||
assert str(conf_name) in _conf_names
|
assert str(conf_name) in _conf_names
|
||||||
|
|
||||||
fn = _conf_fn_w_ext(conf_name)
|
fn = _conf_fn_w_ext(conf_name)
|
||||||
return _config_dir / Path(fn)
|
return get_conf_dir() / Path(fn)
|
||||||
|
|
||||||
|
|
||||||
def repodir() -> Path:
|
def repodir() -> Path:
|
||||||
|
|
@ -271,8 +314,9 @@ def load(
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# create the $HOME/.config/piker dir if dne
|
# create the $HOME/.config/piker dir if dne
|
||||||
if not _config_dir.is_dir():
|
conf_dir: Path = get_conf_dir()
|
||||||
_config_dir.mkdir(
|
if not conf_dir.is_dir():
|
||||||
|
conf_dir.mkdir(
|
||||||
parents=True,
|
parents=True,
|
||||||
exist_ok=True,
|
exist_ok=True,
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,300 @@
|
||||||
|
# piker: trading gear for hackers
|
||||||
|
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||||
|
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU Affero General Public License as published by
|
||||||
|
# the Free Software Foundation, either version 3 of the License, or
|
||||||
|
# (at your option) any later version.
|
||||||
|
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU Affero General Public License for more details.
|
||||||
|
|
||||||
|
# You should have received a copy of the GNU Affero General Public License
|
||||||
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
'''
|
||||||
|
Data-daemon-actor "endpoint-hooks": the service task entry
|
||||||
|
points for `datad.<brokername>`, the per-provider real-time
|
||||||
|
and historical market-data feed daemon.
|
||||||
|
|
||||||
|
The (data vs. broker)d-split mirrors the ep-groupings in
|
||||||
|
`piker.data.validate._eps`: this daemon hosts all `'datad'`
|
||||||
|
eps (live quotes, history loading, symbology search) while
|
||||||
|
its `brokerd.<brokername>` sibling hosts only the live
|
||||||
|
order-control (and thus credentialed) `'brokerd'` eps.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from __future__ import annotations
|
||||||
|
from contextlib import (
|
||||||
|
asynccontextmanager as acm,
|
||||||
|
)
|
||||||
|
from types import ModuleType
|
||||||
|
from typing import (
|
||||||
|
TYPE_CHECKING,
|
||||||
|
AsyncContextManager,
|
||||||
|
)
|
||||||
|
|
||||||
|
import tractor
|
||||||
|
import trio
|
||||||
|
|
||||||
|
from piker.log import (
|
||||||
|
get_logger,
|
||||||
|
get_console_log,
|
||||||
|
)
|
||||||
|
from . import _util
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from .feed import _FeedsBus
|
||||||
|
|
||||||
|
log = get_logger(name=__name__)
|
||||||
|
|
||||||
|
# `datad`-actor-always-enabled mods: the data-side successor
|
||||||
|
# to the old `piker.brokers._daemon._data_mods` set.
|
||||||
|
# NOTE: keeping this list as small as possible is part of
|
||||||
|
# our caps-sec model and should be treated with utmost care!
|
||||||
|
_datad_service_mods: list[str] = [
|
||||||
|
'piker.brokers.core',
|
||||||
|
'piker.brokers.data',
|
||||||
|
'piker.data',
|
||||||
|
'piker.data.feed',
|
||||||
|
'piker.data._sampling',
|
||||||
|
'piker.data._daemon',
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context
|
||||||
|
async def _setup_persistent_datad(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
brokername: str,
|
||||||
|
loglevel: str|None = None,
|
||||||
|
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Allocate an actor-wide service nursery in this
|
||||||
|
`datad.<brokername>` actor such that data-feed tasks
|
||||||
|
(shm writer-samplers, history backfillers, symbology
|
||||||
|
loaders) can be run in the background persistently by
|
||||||
|
the provider backend as needed.
|
||||||
|
|
||||||
|
'''
|
||||||
|
# NOTE: we only need to setup logging once (and only)
|
||||||
|
# here since all hosted daemon tasks will reference
|
||||||
|
# this same log instance's (actor local) state and thus
|
||||||
|
# don't require any further (level) configuration on
|
||||||
|
# their own B)
|
||||||
|
actor: tractor.Actor = tractor.current_actor()
|
||||||
|
tll: str = actor.loglevel
|
||||||
|
log = get_console_log(
|
||||||
|
level=loglevel or tll,
|
||||||
|
name=f'{_util.subsys}.{brokername}',
|
||||||
|
with_tractor_log=bool(tll),
|
||||||
|
)
|
||||||
|
assert log.name == _util.subsys
|
||||||
|
|
||||||
|
from piker.data import feed
|
||||||
|
assert not feed._bus
|
||||||
|
|
||||||
|
# allocate a nursery to the bus for spawning background
|
||||||
|
# tasks which service client IPC requests, normally
|
||||||
|
# `tractor.Context` connections to explicitly required
|
||||||
|
# `datad` endpoints such as:
|
||||||
|
# - `stream_quotes()`,
|
||||||
|
# - `manage_history()`,
|
||||||
|
# - `allocate_persistent_feed()`,
|
||||||
|
# - `open_symbol_search()`
|
||||||
|
# NOTE: see ep invocation details inside `.data.feed`.
|
||||||
|
async with (
|
||||||
|
trio.open_nursery() as service_nursery
|
||||||
|
):
|
||||||
|
bus: _FeedsBus = feed.get_feed_bus(
|
||||||
|
brokername,
|
||||||
|
service_nursery,
|
||||||
|
)
|
||||||
|
assert bus is feed._bus
|
||||||
|
|
||||||
|
# unblock caller
|
||||||
|
await ctx.started()
|
||||||
|
|
||||||
|
# we pin this task to keep the feeds manager active
|
||||||
|
# until the parent actor decides to tear it down
|
||||||
|
await trio.sleep_forever()
|
||||||
|
|
||||||
|
|
||||||
|
def datad_init(
|
||||||
|
brokername: str,
|
||||||
|
loglevel: str|None = None,
|
||||||
|
|
||||||
|
**start_actor_kwargs,
|
||||||
|
|
||||||
|
) -> tuple[
|
||||||
|
ModuleType,
|
||||||
|
dict,
|
||||||
|
AsyncContextManager,
|
||||||
|
]:
|
||||||
|
'''
|
||||||
|
Given an input broker name, load all named arguments
|
||||||
|
which can be passed for daemon endpoint + context spawn
|
||||||
|
as required in every `datad.<brokername>` (actor)
|
||||||
|
service.
|
||||||
|
|
||||||
|
This includes:
|
||||||
|
- load the appropriate <brokername>.py pkg module,
|
||||||
|
- reads any declared `_datad_mods: list[str]` (falling
|
||||||
|
back to the full `__enable_modules__` set for
|
||||||
|
not-yet-split backends) which will be passed to
|
||||||
|
`tractor.ActorNursery.start_actor(enable_modules=)`
|
||||||
|
at actor start time,
|
||||||
|
- deliver a reference to the daemon lifetime fixture,
|
||||||
|
which for now is always the
|
||||||
|
`_setup_persistent_datad()` context defined above.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from piker.brokers import get_brokermod
|
||||||
|
from .validate import get_eps
|
||||||
|
brokermod = get_brokermod(brokername)
|
||||||
|
modpath: str = brokermod.__name__
|
||||||
|
|
||||||
|
# warn (but don't bail) when the backend is missing
|
||||||
|
# some/all of the `datad` ep contract defined by
|
||||||
|
# `piker.data.validate._eps`.
|
||||||
|
datad_eps: dict = get_eps(brokermod, 'datad')
|
||||||
|
if not datad_eps:
|
||||||
|
log.warning(
|
||||||
|
f'Backend {brokername!r} offers NO `datad` '
|
||||||
|
f'(data-feed) eps!?\n'
|
||||||
|
f'Most feed/chart functionality will be '
|
||||||
|
f'broken for this provider..\n'
|
||||||
|
)
|
||||||
|
|
||||||
|
start_actor_kwargs['name'] = f'datad.{brokername}'
|
||||||
|
|
||||||
|
# XXX CRITICAL: include any backend-declared spawn
|
||||||
|
# kwargs, eg. `{'infect_asyncio': True}` required by
|
||||||
|
# `ib`'s embedded `asyncio`-mode `ib_async` usage!
|
||||||
|
start_actor_kwargs.update(
|
||||||
|
getattr(
|
||||||
|
brokermod,
|
||||||
|
'_spawn_kwargs',
|
||||||
|
{},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
# lookup actor-enabled modules declared by the backend
|
||||||
|
# offering the `datad` endpoint(s).
|
||||||
|
enabled: list[str]
|
||||||
|
enabled = start_actor_kwargs['enable_modules'] = [
|
||||||
|
__name__, # so that eps from THIS mod can be invoked
|
||||||
|
modpath,
|
||||||
|
]
|
||||||
|
for submodname in getattr(
|
||||||
|
brokermod,
|
||||||
|
'_datad_mods',
|
||||||
|
# fallback for (flat, less mature) backends which
|
||||||
|
# don't yet declare a daemon-kind mod split.
|
||||||
|
getattr(
|
||||||
|
brokermod,
|
||||||
|
'__enable_modules__',
|
||||||
|
[],
|
||||||
|
),
|
||||||
|
):
|
||||||
|
subpath: str = f'{modpath}.{submodname}'
|
||||||
|
enabled.append(subpath)
|
||||||
|
|
||||||
|
return (
|
||||||
|
brokermod,
|
||||||
|
start_actor_kwargs, # to `ActorNursery.start_actor()`
|
||||||
|
|
||||||
|
# XXX see impl above; contains all (actor global)
|
||||||
|
# setup/teardown expected in all `datad` actor
|
||||||
|
# instances.
|
||||||
|
_setup_persistent_datad,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def spawn_datad(
|
||||||
|
brokername: str,
|
||||||
|
loglevel: str|None = None,
|
||||||
|
|
||||||
|
**tractor_kwargs,
|
||||||
|
|
||||||
|
) -> bool:
|
||||||
|
|
||||||
|
log.info(
|
||||||
|
f'Spawning data-daemon,\n'
|
||||||
|
f'backend: {brokername!r}'
|
||||||
|
)
|
||||||
|
|
||||||
|
(
|
||||||
|
brokermod,
|
||||||
|
tractor_kwargs,
|
||||||
|
daemon_fixture_ep,
|
||||||
|
) = datad_init(
|
||||||
|
brokername,
|
||||||
|
loglevel,
|
||||||
|
**tractor_kwargs,
|
||||||
|
)
|
||||||
|
|
||||||
|
# ask `pikerd` to spawn a new sub-actor and manage it
|
||||||
|
# under its actor nursery
|
||||||
|
from piker.service import Services
|
||||||
|
|
||||||
|
dname: str = tractor_kwargs.pop('name') # f'datad.{brokername}'
|
||||||
|
enable_mods: list[str] = list(dict.fromkeys(
|
||||||
|
_datad_service_mods
|
||||||
|
+
|
||||||
|
tractor_kwargs.pop('enable_modules')
|
||||||
|
))
|
||||||
|
portal = await Services.actor_n.start_actor(
|
||||||
|
dname,
|
||||||
|
enable_modules=enable_mods,
|
||||||
|
debug_mode=Services.debug_mode,
|
||||||
|
**tractor_kwargs,
|
||||||
|
)
|
||||||
|
|
||||||
|
# NOTE: the service mngr expects an already spawned
|
||||||
|
# actor + its portal ref in order to do non-blocking
|
||||||
|
# setup of the `datad` service nursery.
|
||||||
|
await Services.start_service_task(
|
||||||
|
dname,
|
||||||
|
portal,
|
||||||
|
|
||||||
|
# signature of target root-task endpoint
|
||||||
|
daemon_fixture_ep,
|
||||||
|
brokername=brokername,
|
||||||
|
loglevel=loglevel,
|
||||||
|
)
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
@acm
|
||||||
|
async def maybe_spawn_datad(
|
||||||
|
brokername: str,
|
||||||
|
loglevel: str|None = None,
|
||||||
|
|
||||||
|
**pikerd_kwargs,
|
||||||
|
|
||||||
|
) -> tractor.Portal:
|
||||||
|
'''
|
||||||
|
Helper to spawn a datad service *from* a client who
|
||||||
|
wishes to use the sub-actor-daemon but is fine with
|
||||||
|
re-using any existing and contactable `datad`.
|
||||||
|
|
||||||
|
Mas o menos, acts as a cached-actor-getter factory.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from piker.service import maybe_spawn_daemon
|
||||||
|
|
||||||
|
async with maybe_spawn_daemon(
|
||||||
|
service_name=f'datad.{brokername}',
|
||||||
|
service_task_target=spawn_datad,
|
||||||
|
spawn_args={
|
||||||
|
'brokername': brokername,
|
||||||
|
},
|
||||||
|
loglevel=loglevel,
|
||||||
|
|
||||||
|
**pikerd_kwargs,
|
||||||
|
|
||||||
|
) as portal:
|
||||||
|
yield portal
|
||||||
|
|
@ -17,7 +17,7 @@
|
||||||
'''
|
'''
|
||||||
Data feed apis and infra.
|
Data feed apis and infra.
|
||||||
|
|
||||||
This module is enabled for ``brokerd`` daemons and includes mostly
|
This module is enabled for ``datad`` daemons and includes mostly
|
||||||
endpoints and middleware to support our real-time, provider agnostic,
|
endpoints and middleware to support our real-time, provider agnostic,
|
||||||
live market quotes layer. Historical data loading and processing is also
|
live market quotes layer. Historical data loading and processing is also
|
||||||
initiated in parts of the feed bus startup but business logic and
|
initiated in parts of the feed bus startup but business logic and
|
||||||
|
|
@ -54,8 +54,11 @@ from piker.accounting import (
|
||||||
)
|
)
|
||||||
from piker.types import Struct
|
from piker.types import Struct
|
||||||
from piker.brokers import get_brokermod
|
from piker.brokers import get_brokermod
|
||||||
from piker.service import (
|
# NOTE: must be a "relative-direct" import (NOT via
|
||||||
maybe_spawn_brokerd,
|
# `piker.service`) to avoid a partial-init cycle when this
|
||||||
|
# mod is loaded as part of `piker.service.__init__`.
|
||||||
|
from ._daemon import (
|
||||||
|
maybe_spawn_datad,
|
||||||
)
|
)
|
||||||
from piker.calc import humanize
|
from piker.calc import humanize
|
||||||
from ._util import (
|
from ._util import (
|
||||||
|
|
@ -110,7 +113,7 @@ class _FeedsBus(Struct):
|
||||||
'''
|
'''
|
||||||
Data feeds broadcaster and persistence management.
|
Data feeds broadcaster and persistence management.
|
||||||
|
|
||||||
This is a brokerd side api used to manager persistent real-time
|
This is a datad side api used to manager persistent real-time
|
||||||
streams that can be allocated and left alive indefinitely. A bus is
|
streams that can be allocated and left alive indefinitely. A bus is
|
||||||
associated one-to-one with a particular broker backend where the
|
associated one-to-one with a particular broker backend where the
|
||||||
"bus" refers so a multi-symbol bus where quotes are interleaved in
|
"bus" refers so a multi-symbol bus where quotes are interleaved in
|
||||||
|
|
@ -249,7 +252,7 @@ async def allocate_persistent_feed(
|
||||||
'''
|
'''
|
||||||
Create and maintain a "feed bus" which allocates tasks for real-time
|
Create and maintain a "feed bus" which allocates tasks for real-time
|
||||||
streaming and optional historical data storage per broker/data provider
|
streaming and optional historical data storage per broker/data provider
|
||||||
backend; this normally task runs *in* a `brokerd` actor.
|
backend; this normally task runs *in* a `datad` actor.
|
||||||
|
|
||||||
If none exists, this allocates a ``_FeedsBus`` which manages the
|
If none exists, this allocates a ``_FeedsBus`` which manages the
|
||||||
lifetimes of streaming tasks created for each requested symbol.
|
lifetimes of streaming tasks created for each requested symbol.
|
||||||
|
|
@ -318,8 +321,8 @@ async def allocate_persistent_feed(
|
||||||
# at max capacity.
|
# at max capacity.
|
||||||
# - the same ideas ^ but when a local core is maxxed out (like how
|
# - the same ideas ^ but when a local core is maxxed out (like how
|
||||||
# binance does often with hft XD
|
# binance does often with hft XD
|
||||||
# - if a brokerd is non-local then we can't just allocate a mem
|
# - if a datad is non-local then we can't just allocate a mem
|
||||||
# channel here and have the brokerd write it, we instead need
|
# channel here and have the datad write it, we instead need
|
||||||
# a small streaming machine around the remote feed which can then
|
# a small streaming machine around the remote feed which can then
|
||||||
# do the normal work of sampling and writing shm buffers
|
# do the normal work of sampling and writing shm buffers
|
||||||
# (depending on if we want sampling done on the far end or not?)
|
# (depending on if we want sampling done on the far end or not?)
|
||||||
|
|
@ -499,7 +502,7 @@ async def open_feed_bus(
|
||||||
# (after we also group them in a nice `/dev/shm/piker/` subdir).
|
# (after we also group them in a nice `/dev/shm/piker/` subdir).
|
||||||
# ensure we are who we think we are
|
# ensure we are who we think we are
|
||||||
servicename = tractor.current_actor().name
|
servicename = tractor.current_actor().name
|
||||||
assert 'brokerd' in servicename
|
assert 'datad' in servicename
|
||||||
assert brokername in servicename
|
assert brokername in servicename
|
||||||
|
|
||||||
bus: _FeedsBus = get_feed_bus(brokername)
|
bus: _FeedsBus = get_feed_bus(brokername)
|
||||||
|
|
@ -509,12 +512,12 @@ async def open_feed_bus(
|
||||||
for symbol in symbols:
|
for symbol in symbols:
|
||||||
|
|
||||||
# if no cached feed for this symbol has been created for this
|
# if no cached feed for this symbol has been created for this
|
||||||
# brokerd yet, start persistent stream and shm writer task in
|
# datad yet, start persistent stream and shm writer task in
|
||||||
# service nursery
|
# service nursery
|
||||||
flume = bus.feeds.get(symbol)
|
flume = bus.feeds.get(symbol)
|
||||||
if flume is None:
|
if flume is None:
|
||||||
# allocate a new actor-local stream bus which
|
# allocate a new actor-local stream bus which
|
||||||
# will persist for this `brokerd`'s service lifetime.
|
# will persist for this `datad`'s service lifetime.
|
||||||
async with bus.task_lock:
|
async with bus.task_lock:
|
||||||
await bus.nursery.start(
|
await bus.nursery.start(
|
||||||
partial(
|
partial(
|
||||||
|
|
@ -721,7 +724,7 @@ class Feed(Struct):
|
||||||
mods = {name: self.mods[name] for name in brokers}
|
mods = {name: self.mods[name] for name in brokers}
|
||||||
|
|
||||||
if len(mods) == 1:
|
if len(mods) == 1:
|
||||||
# just pass the brokerd stream directly if only one provider
|
# just pass the datad stream directly if only one provider
|
||||||
# was detected.
|
# was detected.
|
||||||
stream = self.streams[list(brokers)[0]]
|
stream = self.streams[list(brokers)[0]]
|
||||||
async with stream.subscribe() as bstream:
|
async with stream.subscribe() as bstream:
|
||||||
|
|
@ -763,7 +766,7 @@ class Feed(Struct):
|
||||||
|
|
||||||
|
|
||||||
@acm
|
@acm
|
||||||
async def install_brokerd_search(
|
async def install_datad_search(
|
||||||
|
|
||||||
portal: tractor.Portal,
|
portal: tractor.Portal,
|
||||||
brokermod: ModuleType,
|
brokermod: ModuleType,
|
||||||
|
|
@ -812,7 +815,7 @@ async def maybe_open_feed(
|
||||||
ReceiveChannel[dict[str, Any]],
|
ReceiveChannel[dict[str, Any]],
|
||||||
):
|
):
|
||||||
'''
|
'''
|
||||||
Maybe open a data to a ``brokerd`` daemon only if there is no
|
Maybe open a data feed to a ``datad`` daemon only if there is no
|
||||||
local one for the broker-symbol pair, if one is cached use it wrapped
|
local one for the broker-symbol pair, if one is cached use it wrapped
|
||||||
in a tractor broadcast receiver.
|
in a tractor broadcast receiver.
|
||||||
|
|
||||||
|
|
@ -885,14 +888,14 @@ async def open_feed(
|
||||||
providers.setdefault(mod, []).append(bs_fqme)
|
providers.setdefault(mod, []).append(bs_fqme)
|
||||||
feed.mods[mod.name] = mod
|
feed.mods[mod.name] = mod
|
||||||
|
|
||||||
# one actor per brokerd for now
|
# one actor per datad for now
|
||||||
brokerd_ctxs = []
|
datad_ctxs = []
|
||||||
for brokermod, bfqmes in providers.items():
|
for brokermod, bfqmes in providers.items():
|
||||||
|
|
||||||
# if no `brokerd` for this backend exists yet we spawn
|
# if no `datad` for this backend exists yet we spawn
|
||||||
# a daemon actor for it.
|
# a daemon actor for it.
|
||||||
brokerd_ctxs.append(
|
datad_ctxs.append(
|
||||||
maybe_spawn_brokerd(
|
maybe_spawn_datad(
|
||||||
brokermod.name,
|
brokermod.name,
|
||||||
loglevel=loglevel
|
loglevel=loglevel
|
||||||
)
|
)
|
||||||
|
|
@ -900,7 +903,7 @@ async def open_feed(
|
||||||
|
|
||||||
portals: tuple[tractor.Portal]
|
portals: tuple[tractor.Portal]
|
||||||
async with trionics.gather_contexts(
|
async with trionics.gather_contexts(
|
||||||
brokerd_ctxs,
|
datad_ctxs,
|
||||||
) as portals:
|
) as portals:
|
||||||
|
|
||||||
bus_ctxs: list[AsyncContextManager] = []
|
bus_ctxs: list[AsyncContextManager] = []
|
||||||
|
|
@ -937,9 +940,9 @@ async def open_feed(
|
||||||
tick_throttle=tick_throttle,
|
tick_throttle=tick_throttle,
|
||||||
|
|
||||||
# XXX: super important to avoid
|
# XXX: super important to avoid
|
||||||
# the brokerd from some other
|
# the datad from some other
|
||||||
# backend overruning the task here
|
# backend overruning the task here
|
||||||
# bc some other brokerd took longer
|
# bc some other datad took longer
|
||||||
# to startup before we hit the `.open_stream()`
|
# to startup before we hit the `.open_stream()`
|
||||||
# loop below XD .. really we should try to do each
|
# loop below XD .. really we should try to do each
|
||||||
# of these stream open sequences sequentially per
|
# of these stream open sequences sequentially per
|
||||||
|
|
@ -1008,7 +1011,7 @@ async def open_feed(
|
||||||
assert stream
|
assert stream
|
||||||
feed.streams[brokermod.name] = stream
|
feed.streams[brokermod.name] = stream
|
||||||
|
|
||||||
# apply `brokerd`-common stream to each flume
|
# apply `datad`-common stream to each flume
|
||||||
# tracking a live market feed from that provider.
|
# tracking a live market feed from that provider.
|
||||||
for fqme, flume in feed.flumes.items():
|
for fqme, flume in feed.flumes.items():
|
||||||
if brokermod.name == flume.mkt.broker:
|
if brokermod.name == flume.mkt.broker:
|
||||||
|
|
|
||||||
|
|
@ -91,6 +91,24 @@ _eps: dict[str, list[str]] = {
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_eps(
|
||||||
|
mod: ModuleType,
|
||||||
|
kind: str, # 'middleware' | 'datad' | 'brokerd'
|
||||||
|
|
||||||
|
) -> dict[str, Callable]:
|
||||||
|
'''
|
||||||
|
Return the daemon-kind's ep funcs defined by the backend
|
||||||
|
`mod`, keyed by ep name; any eps from `_eps[kind]` not
|
||||||
|
implemented by the backend are excluded.
|
||||||
|
|
||||||
|
'''
|
||||||
|
return {
|
||||||
|
name: ep
|
||||||
|
for name in _eps[kind]
|
||||||
|
if (ep := getattr(mod, name, None))
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def validate_backend(
|
def validate_backend(
|
||||||
mod: ModuleType,
|
mod: ModuleType,
|
||||||
syms: list[str],
|
syms: list[str],
|
||||||
|
|
|
||||||
|
|
@ -56,3 +56,7 @@ from ..brokers._daemon import (
|
||||||
spawn_brokerd as spawn_brokerd,
|
spawn_brokerd as spawn_brokerd,
|
||||||
maybe_spawn_brokerd as maybe_spawn_brokerd,
|
maybe_spawn_brokerd as maybe_spawn_brokerd,
|
||||||
)
|
)
|
||||||
|
from ..data._daemon import (
|
||||||
|
spawn_datad as spawn_datad,
|
||||||
|
maybe_spawn_datad as maybe_spawn_datad,
|
||||||
|
)
|
||||||
|
|
|
||||||
|
|
@ -157,6 +157,7 @@ _root_modules: list[str] = [
|
||||||
__name__,
|
__name__,
|
||||||
'piker.service._daemon',
|
'piker.service._daemon',
|
||||||
'piker.brokers._daemon',
|
'piker.brokers._daemon',
|
||||||
|
'piker.data._daemon',
|
||||||
|
|
||||||
'piker.clearing._ems',
|
'piker.clearing._ems',
|
||||||
'piker.clearing._client',
|
'piker.clearing._client',
|
||||||
|
|
@ -214,7 +215,13 @@ async def open_pikerd(
|
||||||
trio.open_nursery() as service_tn,
|
trio.open_nursery() as service_tn,
|
||||||
):
|
):
|
||||||
for addr in reg_addrs:
|
for addr in reg_addrs:
|
||||||
uaddr: tuple = addr.unwrap()
|
# normalize to a wrapped `tractor` addr-type;
|
||||||
|
# entries may be raw `tuple`s when passed in
|
||||||
|
# from (test) client code.
|
||||||
|
wladdr = tractor.discovery._addr.wrap_address(
|
||||||
|
addr,
|
||||||
|
)
|
||||||
|
uaddr: tuple = wladdr.unwrap()
|
||||||
if (
|
if (
|
||||||
uaddr not in root_actor.accept_addrs
|
uaddr not in root_actor.accept_addrs
|
||||||
):
|
):
|
||||||
|
|
|
||||||
|
|
@ -225,10 +225,13 @@ async def check_for_service(
|
||||||
|
|
||||||
'''
|
'''
|
||||||
async with (
|
async with (
|
||||||
open_registry(ensure_exists=False) as reg_addr,
|
open_registry(
|
||||||
|
addrs=Registry.addrs,
|
||||||
|
ensure_exists=False,
|
||||||
|
) as reg_addrs,
|
||||||
tractor.query_actor(
|
tractor.query_actor(
|
||||||
service_name,
|
service_name,
|
||||||
arbiter_sockaddr=reg_addr,
|
regaddr=reg_addrs[0],
|
||||||
) as sockaddr,
|
) as (sockaddr, _),
|
||||||
):
|
):
|
||||||
return sockaddr
|
return sockaddr
|
||||||
|
|
|
||||||
|
|
@ -32,12 +32,12 @@ from . import _event
|
||||||
from . import _search
|
from . import _search
|
||||||
from ..accounting import unpack_fqme
|
from ..accounting import unpack_fqme
|
||||||
from ..data._symcache import open_symcache
|
from ..data._symcache import open_symcache
|
||||||
from ..data.feed import install_brokerd_search
|
from ..data.feed import install_datad_search
|
||||||
from ..log import (
|
from ..log import (
|
||||||
get_logger,
|
get_logger,
|
||||||
get_console_log,
|
get_console_log,
|
||||||
)
|
)
|
||||||
from ..service import maybe_spawn_brokerd
|
from ..service import maybe_spawn_datad
|
||||||
from ._exec import run_qtractor
|
from ._exec import run_qtractor
|
||||||
|
|
||||||
log = get_logger(__name__)
|
log = get_logger(__name__)
|
||||||
|
|
@ -50,16 +50,16 @@ async def load_provider_search(
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|
||||||
name = brokermod.name
|
name = brokermod.name
|
||||||
log.info(f'loading brokerd for {name}..')
|
log.info(f'loading datad for {name}..')
|
||||||
|
|
||||||
async with (
|
async with (
|
||||||
|
|
||||||
maybe_spawn_brokerd(
|
maybe_spawn_datad(
|
||||||
name,
|
name,
|
||||||
loglevel=loglevel
|
loglevel=loglevel
|
||||||
) as portal,
|
) as portal,
|
||||||
|
|
||||||
install_brokerd_search(
|
install_datad_search(
|
||||||
portal,
|
portal,
|
||||||
brokermod,
|
brokermod,
|
||||||
),
|
),
|
||||||
|
|
|
||||||
|
|
@ -27,7 +27,7 @@ from ..cli import (
|
||||||
load_trans_eps,
|
load_trans_eps,
|
||||||
)
|
)
|
||||||
from .. import watchlists as wl
|
from .. import watchlists as wl
|
||||||
from ..service import maybe_spawn_brokerd
|
from ..service import maybe_spawn_datad
|
||||||
|
|
||||||
|
|
||||||
_config_dir = click.get_app_dir('piker')
|
_config_dir = click.get_app_dir('piker')
|
||||||
|
|
@ -69,7 +69,7 @@ def monitor(config, rate, name, dhost, test, tl):
|
||||||
from .kivy.monitor import _async_main
|
from .kivy.monitor import _async_main
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
async with maybe_spawn_brokerd(
|
async with maybe_spawn_datad(
|
||||||
brokername=brokermod.name,
|
brokername=brokermod.name,
|
||||||
loglevel=loglevel
|
loglevel=loglevel
|
||||||
) as portal:
|
) as portal:
|
||||||
|
|
@ -118,7 +118,7 @@ def optschain(
|
||||||
from .kivy.option_chain import _async_main
|
from .kivy.option_chain import _async_main
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
async with maybe_spawn_brokerd(
|
async with maybe_spawn_datad(
|
||||||
loglevel=loglevel
|
loglevel=loglevel
|
||||||
):
|
):
|
||||||
# run app "main"
|
# run app "main"
|
||||||
|
|
|
||||||
|
|
@ -495,7 +495,7 @@ async def _async_main(
|
||||||
|
|
||||||
async with trio.open_nursery() as nursery:
|
async with trio.open_nursery() as nursery:
|
||||||
# get a portal to the data feed daemon
|
# get a portal to the data feed daemon
|
||||||
async with tractor.wait_for_actor('brokerd') as portal:
|
async with tractor.wait_for_actor('datad') as portal:
|
||||||
|
|
||||||
# set up a pager view for large ticker lists
|
# set up a pager view for large ticker lists
|
||||||
chain = await new_chain_ui(
|
chain = await new_chain_ui(
|
||||||
|
|
|
||||||
|
|
@ -151,7 +151,7 @@ def load_and_check_pos(
|
||||||
# is the same the fqme.
|
# is the same the fqme.
|
||||||
pp: Position = table.pps[ppmsg.symbol]
|
pp: Position = table.pps[ppmsg.symbol]
|
||||||
|
|
||||||
assert ppmsg.size == pp.size
|
assert ppmsg.size == pp.cumsize
|
||||||
assert ppmsg.avg_price == pp.ppu
|
assert ppmsg.avg_price == pp.ppu
|
||||||
|
|
||||||
yield pp
|
yield pp
|
||||||
|
|
@ -179,7 +179,7 @@ def test_ems_err_on_bad_broker(
|
||||||
# NOTE: emsd should error on the actor's enabled modules
|
# NOTE: emsd should error on the actor's enabled modules
|
||||||
# import phase, when looking for a backend named `doggy`.
|
# import phase, when looking for a backend named `doggy`.
|
||||||
except tractor.RemoteActorError as re:
|
except tractor.RemoteActorError as re:
|
||||||
assert re.type is ModuleNotFoundError
|
assert re.boxed_type is ModuleNotFoundError
|
||||||
|
|
||||||
run_and_tollerate_cancels(load_bad_fqme)
|
run_and_tollerate_cancels(load_bad_fqme)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -53,11 +53,12 @@ def test_runtime_boot(
|
||||||
|
|
||||||
tractor.wait_for_actor(
|
tractor.wait_for_actor(
|
||||||
'pikerd',
|
'pikerd',
|
||||||
arbiter_sockaddr=daemon_addr,
|
registry_addr=daemon_addr,
|
||||||
) as portal,
|
) as portal,
|
||||||
):
|
):
|
||||||
assert pikerd_portal.channel.raddr == daemon_addr
|
uw_raddr: tuple = pikerd_portal.chan.raddr.unwrap()
|
||||||
assert pikerd_portal.channel.raddr == portal.channel.raddr
|
assert uw_raddr == daemon_addr
|
||||||
|
assert uw_raddr == portal.chan.raddr.unwrap()
|
||||||
|
|
||||||
# no service tasks should be started
|
# no service tasks should be started
|
||||||
assert not services.service_tasks
|
assert not services.service_tasks
|
||||||
|
|
@ -65,6 +66,41 @@ def test_runtime_boot(
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
|
def test_datad_spawn(
|
||||||
|
open_test_pikerd: AsyncContextManager,
|
||||||
|
loglevel: str,
|
||||||
|
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Verify the new (data-feed-only) `datad.<broker>` daemon
|
||||||
|
can be spawned/registered as a `pikerd` sub-service via
|
||||||
|
the `maybe_spawn_datad()` factory.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from piker.service import maybe_spawn_datad
|
||||||
|
|
||||||
|
backend: str = 'kraken'
|
||||||
|
datad_name: str = f'datad.{backend}'
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
async with (
|
||||||
|
open_test_pikerd() as (_, _, _, services),
|
||||||
|
|
||||||
|
maybe_spawn_datad(
|
||||||
|
backend,
|
||||||
|
loglevel=loglevel,
|
||||||
|
) as portal,
|
||||||
|
):
|
||||||
|
assert portal
|
||||||
|
|
||||||
|
async with ensure_service(datad_name):
|
||||||
|
assert (
|
||||||
|
datad_name in services.service_tasks
|
||||||
|
)
|
||||||
|
|
||||||
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
def test_ensure_datafeed_actors(
|
def test_ensure_datafeed_actors(
|
||||||
open_test_pikerd: AsyncContextManager,
|
open_test_pikerd: AsyncContextManager,
|
||||||
loglevel: str,
|
loglevel: str,
|
||||||
|
|
@ -72,14 +108,14 @@ def test_ensure_datafeed_actors(
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
'''
|
'''
|
||||||
Verify that booting a data feed starts a `brokerd`
|
Verify that booting a data feed starts a `datad`
|
||||||
actor and a singleton global `samplerd` and opening
|
actor and a singleton global `samplerd` and opening
|
||||||
an order mode in paper opens the `paperboi` service.
|
an order mode in paper opens the `paperboi` service.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
actor_name: str = 'brokerd'
|
actor_name: str = 'datad'
|
||||||
backend: str = 'kraken'
|
backend: str = 'kraken'
|
||||||
brokerd_name: str = f'{actor_name}.{backend}'
|
datad_name: str = f'{actor_name}.{backend}'
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
async with (
|
async with (
|
||||||
|
|
@ -94,7 +130,7 @@ def test_ensure_datafeed_actors(
|
||||||
await feed.pause()
|
await feed.pause()
|
||||||
|
|
||||||
async with (
|
async with (
|
||||||
ensure_service(brokerd_name),
|
ensure_service(datad_name),
|
||||||
ensure_service('samplerd'),
|
ensure_service('samplerd'),
|
||||||
):
|
):
|
||||||
await trio.sleep(0.1)
|
await trio.sleep(0.1)
|
||||||
|
|
@ -108,7 +144,7 @@ async def ensure_service(
|
||||||
sockaddr: tuple[str, int] | None = None,
|
sockaddr: tuple[str, int] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
async with find_service(name) as portal:
|
async with find_service(name) as portal:
|
||||||
remote_sockaddr = portal.channel.raddr
|
remote_sockaddr: tuple = portal.chan.raddr.unwrap()
|
||||||
print(f'FOUND `{name}` @ {remote_sockaddr}')
|
print(f'FOUND `{name}` @ {remote_sockaddr}')
|
||||||
|
|
||||||
if sockaddr:
|
if sockaddr:
|
||||||
|
|
@ -131,40 +167,49 @@ def run_test_w_cancel_method(
|
||||||
"was remotely cancelled by remote actor (\'pikerd\'")
|
"was remotely cancelled by remote actor (\'pikerd\'")
|
||||||
|
|
||||||
if cancel_method == 'sigint':
|
if cancel_method == 'sigint':
|
||||||
with pytest.raises(
|
# XXX: with modern `tractor` the (single-exc)
|
||||||
|
# group is collapsed so a bare KBI normally
|
||||||
|
# propagates; tolerate either form.
|
||||||
|
with pytest.raises((
|
||||||
|
KeyboardInterrupt,
|
||||||
BaseExceptionGroup,
|
BaseExceptionGroup,
|
||||||
) as exc_info:
|
)) as exc_info:
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
multi = exc_info.value
|
err = exc_info.value
|
||||||
|
match err:
|
||||||
|
case BaseExceptionGroup():
|
||||||
|
for suberr in err.exceptions:
|
||||||
|
match suberr:
|
||||||
|
# ensure we receive a remote
|
||||||
|
# cancellation error caused by the
|
||||||
|
# pikerd root actor.
|
||||||
|
case tractor.ContextCancelled():
|
||||||
|
assert (
|
||||||
|
cancelled_msg
|
||||||
|
in
|
||||||
|
suberr.args[0]
|
||||||
|
)
|
||||||
|
|
||||||
for suberr in multi.exceptions:
|
case KeyboardInterrupt():
|
||||||
match suberr:
|
pass
|
||||||
# ensure we receive a remote cancellation error caused
|
|
||||||
# by the pikerd root actor since we used the
|
|
||||||
# `.cancel_service()` API above B)
|
|
||||||
case tractor.ContextCancelled():
|
|
||||||
assert cancelled_msg in suberr.args[0]
|
|
||||||
|
|
||||||
case KeyboardInterrupt():
|
case _:
|
||||||
pass
|
pytest.fail(
|
||||||
|
f'Unexpected error {suberr}'
|
||||||
|
)
|
||||||
|
|
||||||
case _:
|
case KeyboardInterrupt():
|
||||||
pytest.fail(f'Unexpected error {suberr}')
|
pass
|
||||||
|
|
||||||
elif cancel_method == 'services':
|
elif cancel_method == 'services':
|
||||||
|
# XXX: cancelling our own sub-service via
|
||||||
# XXX NOTE: oddly, when you pass --pdb to pytest, i think since
|
# `Services.cancel_service()` is a *self*
|
||||||
# we also use that to enable the underlying tractor debug mode,
|
# requested cancel: modern `tractor` absorbs the
|
||||||
# it causes this to not raise for some reason? So if you see
|
# resulting `ContextCancelled` (canceller is our
|
||||||
# that while changing this test.. it's prolly that.
|
# own actor) so the runtime tears down gracefully
|
||||||
|
# with NO error raised to the opener.
|
||||||
with pytest.raises(
|
trio.run(main)
|
||||||
tractor.ContextCancelled
|
|
||||||
) as exc_info:
|
|
||||||
trio.run(main)
|
|
||||||
|
|
||||||
assert cancelled_msg in exc_info.value.args[0]
|
|
||||||
|
|
||||||
else:
|
else:
|
||||||
pytest.fail(f'Test is broken due to {cancel_method}')
|
pytest.fail(f'Test is broken due to {cancel_method}')
|
||||||
|
|
@ -182,9 +227,9 @@ def test_ensure_ems_in_paper_actors(
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|
||||||
actor_name: str = 'brokerd'
|
|
||||||
backend: str = 'kraken'
|
backend: str = 'kraken'
|
||||||
brokerd_name: str = f'{actor_name}.{backend}'
|
datad_name: str = f'datad.{backend}'
|
||||||
|
brokerd_name: str = f'brokerd.{backend}'
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
|
|
||||||
|
|
@ -197,7 +242,9 @@ def test_ensure_ems_in_paper_actors(
|
||||||
# ensure we timeout after is startup is too slow.
|
# ensure we timeout after is startup is too slow.
|
||||||
# TODO: something like this should be our start point for
|
# TODO: something like this should be our start point for
|
||||||
# benchmarking end-to-end startup B)
|
# benchmarking end-to-end startup B)
|
||||||
with trio.fail_after(9):
|
# NOTE: includes a live (kraken) symbology fetch so
|
||||||
|
# the budget needs some headroom for net latency..
|
||||||
|
with trio.fail_after(19):
|
||||||
async with (
|
async with (
|
||||||
open_test_pikerd() as (_, _, _, services),
|
open_test_pikerd() as (_, _, _, services),
|
||||||
|
|
||||||
|
|
@ -226,15 +273,25 @@ def test_ensure_ems_in_paper_actors(
|
||||||
|
|
||||||
async with (
|
async with (
|
||||||
ensure_service('emsd'),
|
ensure_service('emsd'),
|
||||||
ensure_service(brokerd_name),
|
ensure_service(datad_name),
|
||||||
ensure_service(f'paperboi.{backend}'),
|
ensure_service(f'paperboi.{backend}'),
|
||||||
):
|
):
|
||||||
for name in pikerd_subservices:
|
for name in pikerd_subservices:
|
||||||
assert name in services.service_tasks
|
assert name in services.service_tasks
|
||||||
|
|
||||||
# brokerd.kraken actor should have been started
|
# datad.kraken actor should have been
|
||||||
# implicitly by the ems.
|
# started implicitly by the feed layer.
|
||||||
assert brokerd_name in services.service_tasks
|
assert datad_name in services.service_tasks
|
||||||
|
|
||||||
|
# XXX: paper-mode sessions should NEVER
|
||||||
|
# boot a (live, credentialed) `brokerd`;
|
||||||
|
# only emsd's `open_brokerd_dialog()`
|
||||||
|
# live-ep path is allowed to spawn it!
|
||||||
|
assert (
|
||||||
|
brokerd_name
|
||||||
|
not in
|
||||||
|
services.service_tasks
|
||||||
|
)
|
||||||
|
|
||||||
print('ALL SERVICES STARTED, cancelling runtime with:\n'
|
print('ALL SERVICES STARTED, cancelling runtime with:\n'
|
||||||
f'-> {cancel_method}')
|
f'-> {cancel_method}')
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue