piker/ai/claude-code/plans/datad_service.md

Split brokerd.<broker> into trading-only brokerd + new datad.<broker>

Context

Today a single brokerd.<broker> actor hosts BOTH concerns:

• data feed service tasks: the _FeedsBus + open_feed_bus() ep (piker/data/feed.py:464), per-symbol allocate_persistent_feed() tasks (shm writers via sample_and_broadcast(), history backfill via piker.tsp), plus backend eps stream_quotes, open_history_client, open_symbol_search, get_mkt_info from piker/brokers/<backend>/feed.py/symbols.py,
• live order-control tasks: open_trade_dialog() from piker/brokers/<backend>/broker.py, driven by emsd.

The codebase already anticipates this split: piker/data/validate.py:70-91 groups backend eps into 'datad' vs 'brokerd' kinds, piker/brokers/ib/__init__.py:62-70 already declares _brokerd_mods/_datad_mods, and piker/brokers/_daemon.py:62 carries the literal TODO “rename the daemon to datad prolly once we split up broker vs. data tasks into separate actors?”. This work executes that split.

User-decided constraints:

• datad.<broker> is a sibling of brokerd.<broker> under pikerd, spawned via the existing Services + maybe_spawn_daemon() machinery (piker/service/_daemon.py:46).
• Hard cutover, staged by layer — no dual-mode runtime flag; every stage lands fully working.
• Post-split brokerd is trading-only and lazily spawned solely by emsd’s open_brokerd_dialog path; UI/CLI/feed code never spawns it. Chart-only and paper sessions run with zero brokerd processes.

Target topology

pikerd
├── datad.ib      feed bus, shm writers, tsp history, symbol search
├── brokerd.ib    open_trade_dialog only (EMS-spawned, lazy)
├── emsd
│   └── paperboi.ib   (paper mode; opens its feed via datad)
└── samplerd

Key verified facts (load-bearing)

• feed.portals has exactly ONE trading consumer: piker/clearing/_ems.py:671 (portal = feed.portals[brokermod] → handed to open_brokerd_dialog). This is the single coupling forcing feed + trading into one actor.
• assert 'brokerd' in servicename at piker/data/feed.py:502 is the only actor-name assert in the tree.
• piker ledger (piker/accounting/cli.py:100-157) is a hidden consumer: it calls broker_init() directly, spawns its own ad-hoc actor, enters _setup_persistent_brokerd, then calls open_brokerd_dialog(brokermod, portal, ...) with an explicit portal — the new signature must keep a portal: override param.
• kraken’s feed→broker coupling is mild: kraken/broker.py:77-81 imports NoBsWs/open_autorecon_ws (really from piker.data._web_bs) and stream_messages (pure parser) from feed.py. In-process imports only — enable_modules gates RPC, not imports. No live shared state; each actor opens its own WS.
• ib: default client_id=6116 (ib/api.py:1320) with linear retry clientId=client_id + i (:1403). Two ib actors connecting concurrently will collide → needs a role-based id offset. Both daemons need _spawn_kwargs={'infect_asyncio': True} (copied by both init fns).
• binance has no symbols.py (get_mkt_info/open_symbol_search live in binance/feed.py); kucoin/questrade/robinhood are flat single-file with NO __enable_modules__; deribit has no broker.py.
• tests/test_services.py:80,185 assert brokerd actor names incl. the paper-mode flow asserting brokerd.kraken spawns (:229,:237) — must invert post-split.
• _root_modules (piker/service/_actor_runtime.py:156) must gain the new datad daemon mod so pikerd_portal.run(spawn_datad, ...) resolves.
• piker/brokers/core.py:175 symbol_search does portal.run(search_w_brokerd) where the target fn lives in piker.brokers.core → that mod must stay RPC-enabled in datad.

Module decision

New file piker/data/_daemon.py hosts all datad machinery (_setup_persistent_datad, datad_init, spawn_datad, maybe_spawn_datad, _datad_service_mods). Mirrors the piker.brokers._daemon / piker.data._sampling (samplerd) per-subsystem convention and satisfies the existing TODO at brokers/_daemon.py:49 (“move this def to the .data subpkg”). piker.brokers._daemon keeps brokerd-only code, slimmed. Do NOT over-DRY the two ~40-line init fns into a shared factory yet (samplerd precedent accepts the duplication).

---

Stage 0 — prep: backend module grouping + ep introspection (no topology change)

1. piker/data/validate.py: add a get_eps(mod, kind) -> dict[str, Callable] helper returning the backend’s defined eps for a daemon-kind from _eps (missing eps excluded). Used later by both init fns + fail-fast checks.
2. Declare daemon-kind module groups in each split backend’s __init__.py, keeping __enable_modules__ as the (deduped) union so behavior is unchanged:
   • kraken: _brokerd_mods = ['api', 'broker'], _datad_mods = ['api', 'feed', 'symbols']
   • binance: _brokerd_mods = ['api', 'broker'], _datad_mods = ['api', 'feed']
   • deribit: _brokerd_mods = [], _datad_mods = ['api', 'feed']
   • ib: adjust existing groups so each includes 'api'
   • flat backends (kucoin etc.): no attrs — init fns fall back to enabling just modpath (existing behavior).
3. Hygiene: kraken/broker.py:77-81 imports NoBsWs/open_autorecon_ws from piker.data._web_bs directly (keep the stream_messages import from .feed).

Gate: full pytest green, zero behavior change.

Stage 1 — introduce datad daemon machinery (additive)

New piker/data/_daemon.py:

• _datad_service_mods: list[str] — datad-always-enabled mods (successor to the data side of _data_mods from brokers/_daemon.py:52): ['piker.brokers.core', 'piker.brokers.data', 'piker.data', 'piker.data.feed', 'piker.data._sampling', 'piker.data._daemon'].
• _setup_persistent_datad(ctx, brokername, loglevel) — @tractor.context fixture; logging boilerplate (as _setup_persistent_brokerd:81-88), then allocates the actor-global feed bus exactly as the brokerd fixture does today (brokers/_daemon.py:105-121): assert not feed._bus, open service nursery, feed.get_feed_bus(brokername, service_nursery), ctx.started(), sleep_forever().
• datad_init(brokername, ...) — mirrors broker_init() (brokers/_daemon.py:132): actor name f'datad.{brokername}', copies backend _spawn_kwargs (critical for ib infect_asyncio), builds enable_modules from getattr(brokermod, '_datad_mods', getattr(brokermod, '__enable_modules__', [])).
• spawn_datad(brokername, ...) — mirrors spawn_brokerd() (brokers/_daemon.py:202): Services.actor_n.start_actor(dname, enable_modules=_datad_service_mods + backend_mods, ...) + Services.start_service_task(dname, portal, _setup_persistent_datad, ...).
• maybe_spawn_datad(brokername, ...) — wraps maybe_spawn_daemon(service_name=f'datad.{brokername}', service_task_target=spawn_datad, ...) exactly like maybe_spawn_brokerd (brokers/_daemon.py:256).

Supporting edits:

• piker/service/_actor_runtime.py:156: add 'piker.data._daemon' to _root_modules (keep 'piker.brokers._daemon').
• piker/service/__init__.py: re-export spawn_datad/maybe_spawn_datad next to the brokerd ones (:56-57).
• tests/test_services.py: new test_datad_spawn — open_test_pikerd + maybe_spawn_datad('kraken') + ensure_service('datad.kraken'). (Do NOT route a feed through it yet — open_feed_bus’s assert still says brokerd.)

Gate: suite green + new spawn test; nothing routes through datad yet.

Stage 2 — clearing layer: emsd self-spawns brokerd (decouple from feed.portals)

Sequenced BEFORE the feed cutover so live trading works at every boundary (otherwise feed.portals would hand emsd a datad portal).

piker/clearing/_ems.py:

1. open_brokerd_dialog() (:336) new signature: (brokermod, exec_mode, fqme=None, portal: tractor.Portal|None = None, loglevel=None). Internals: keep trades-ep detection (:400-417); acquire the brokerd actor ONLY when a live ep will actually open, via a small inner @acm _acquire_live_portal() that yields the passed portal if given (the piker ledger path) else async with maybe_spawn_brokerd(brokermod.name, loglevel=loglevel) — the single place brokerd boots post-split. Move the eager portal.open_context(trades_endpoint, ...) construction (:425) inside that block; the paper short-circuit (:432) never touches it.
2. Router.maybe_open_brokerd_dialog() (:581) — drop the portal param; Router.open_trade_relays() (:640) — delete portal = feed.portals[brokermod] (:671) and the pass-through (:685).
3. piker/accounting/cli.py:144: switch to keyword form open_brokerd_dialog(brokermod, exec_mode=..., portal=portal, loglevel=...).

Pre-cutover this is a pure refactor: emsd’s maybe_spawn_brokerd just finds the already-running brokerd via the registry.

Gate: tests/test_ems.py + services suite green; manual paper order on kraken; piker ledger smoke.

Stage 3 — feed layer hard cutover to datad

1. piker/data/feed.py: import maybe_spawn_datad (replacing maybe_spawn_brokerd, :56-58); open_feed() :895 → maybe_spawn_datad(...) (rename brokerd_ctxs → datad_ctxs); open_feed_bus() :502 → assert 'datad' in servicename; comment sweep.
2. piker/brokers/_daemon.py _setup_persistent_brokerd(): slim to trading-only fixture — logging setup, ctx.started(), sleep_forever(). Drop the bus alloc, assert not feed._bus, the service nursery (backend open_trade_dialog ctxs own their task trees), the eg.ExceptionGroup handler, and the stale _FeedsBus TYPE_CHECKING import. (piker ledger’s ad-hoc actor enters this same slimmed fixture — exactly what it needs.)
3. Repoint remaining data-flavored spawn sites maybe_spawn_brokerd → maybe_spawn_datad:
   • piker/ui/_app.py:40,57 (symbol search; optionally rename install_brokerd_search → install_datad_search, def at piker/data/feed.py:766)
   • piker/brokers/core.py:33,175 (symbol_search)
   • piker/brokers/cli.py:38,190,320 (brokercheck, record)
   • piker/ui/cli.py:30,72,121 (legacy kivy monitor/optschain) + best-effort piker/ui/kivy/option_chain.py:498 wait_for_actor('brokerd') → 'datad'
4. tests/test_services.py: :80,:185 actor_name = 'datad'; paper-mode test asserts datad.kraken + paperboi.kraken + emsd AND adds the headline negative check: assert 'brokerd.kraken' not in services.service_tasks.

Gate: full suite with inverted assertions; manual: chart on binance/kraken shows datad.<broker> in piker services and NO brokerd; paper order fills; symbol search works; history backfill sane.

Stage 4 — caps-sec slimming, validation, ib client-id

1. piker/brokers/_daemon.py: delete _data_mods (:52); spawn_brokerd() :236 → enable_modules=tractor_kwargs.pop('enable_modules'); broker_init() :184 reads getattr(brokermod, '_brokerd_mods', getattr(brokermod, '__enable_modules__', [])). Resulting brokerd enable set has no piker.data.* at all.
2. Fail-fast ep validation via Stage-0 validate.get_eps(): broker_init raises a clear error when get_eps(mod, 'brokerd') is empty (e.g. “kucoin is datad-only — use paper mode”); datad_init warns analogously.
3. ib client-id mitigation in load_aio_clients() (ib/api.py:~1316): role-based offset when client_id is the 6116 default (e.g. +16 when 'datad' in tractor.current_actor().name), optionally configurable via brokers.toml keys.
4. Docs/comment sweep: resolve brokers/_daemon.py:62 TODO, piker/cli/__init__.py:253 TODO, daemon list in piker/service/__init__.py:21 docstring.

Gate: full suite; audit greps clean: grep -rn maybe_spawn_brokerd piker/ hits only clearing/_ems.py, service/__init__.py, brokers/_daemon.py; no spawn-path 'brokerd' literals left in piker/data, piker/ui, piker/brokers/cli.py.

---

Risk register

Risk	Sev	Mitigation
ib: datad.ib + brokerd.ib both connect to TWS/gw; client_id=6116 collision burns connect_timeout retries	High (live ib only)	Stage 4 role-based id offset; both daemons inherit infect_asyncio via _spawn_kwargs copy in both init fns
piker ledger ad-hoc actor path silently broken by signature change	Med	portal: override kept on open_brokerd_dialog (Stage 2); explicit smoke test
datad-only backends (kucoin/deribit) → useless brokerd spawn	Low	already covered: open_brokerd_dialog forces exec_mode='paper' when no trades ep (_ems.py:413-417) BEFORE the lazy spawn; Stage 4 fail-fast
brokerd in-process symbology/ledger needs (ib broker.py imports .symbols; kraken uses stream_messages)	None	verified pure in-process imports; enable_modules gates RPC only
paper-mode test asserts brokerd spawns (test_services.py:237)	Low	Stage 3 deliberately inverts it
doubled per-broker clients (HTTP/WS, symcache loads)	Low	each actor already opens its own WS today; symcache is disk-read-mostly

Verification (per stage gates above, plus end-to-end)

• pytest tests/test_services.py tests/test_feeds.py tests/test_ems.py at every stage (kraken/binance public endpoints, no creds needed).
• Manual smoke matrix post-Stage-3: pikerd -l info + piker chart btcusdt.spot.binance → piker services shows datad.binance, no brokerd; submit paper order → paperboi.binance appears, still no brokerd; symbol search; piker ledger <broker>.paper.
• If ib creds/gateway available: live ib chart + small live order to exercise dual-actor client-id path (Stage 4).

Critical files

• new piker/data/_daemon.py — all datad machinery
• piker/brokers/_daemon.py — slimmed brokerd fixture/init/spawn
• piker/data/feed.py — spawn cutover + actor-name assert
• piker/clearing/_ems.py — emsd lazy brokerd spawn (open_brokerd_dialog, Router)
• piker/service/_actor_runtime.py, piker/service/__init__.py — root mods + re-exports
• piker/data/validate.py — get_eps() helper
• backend __init__.pys (kraken/binance/deribit/ib) — _datad_mods/_brokerd_mods
• piker/accounting/cli.py — keyword-form dialog call
• tests/test_services.py — inverted + new assertions