Compare commits
No commits in common. "0404e4230ef00cbde21cfa2f73006b4dd1211fcd" and "f084e89991f2e1e941c0b8abe2899d6b7506238f" have entirely different histories.
0404e4230e
...
f084e89991
|
|
@ -1,60 +0,0 @@
|
|||
---
|
||||
model: claude-fable-5[1m]
|
||||
service: claude
|
||||
session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
|
||||
timestamp: 2026-06-10T21:35:49Z
|
||||
git_ref: datad_service
|
||||
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.md
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260610T213549Z_f084e899_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
User bug report from live `ib` testing of the
|
||||
(datad|brokerd)-split branch:
|
||||
|
||||
> ok doing some testing and noticing live orders do not
|
||||
> work otb, pikerd show this on live submission,
|
||||
> [pdb traceback: `RuntimeError("Can not order
|
||||
> {symbol}, no live feed?")` at `ib/api.py:1152`
|
||||
> `submit_limit()` with `self._contracts == {}` in the
|
||||
> `brokerd.ib` actor + a `TrioTaskExited` teardown
|
||||
> cascade]
|
||||
|
||||
Follow-up user direction fixing the design intent:
|
||||
|
||||
> i think the main thing here is that [brokerd] should
|
||||
> be able to always submit orders without a live feed
|
||||
> being up and registered in the same subactor right?
|
||||
|
||||
## Response summary
|
||||
|
||||
Root cause: pre-split the feed code populated the
|
||||
actor-local `Client._contracts` cache (via
|
||||
`get_mkt_info()` writing `mkt.bs_fqme` keys) in the
|
||||
SAME process as order submission; post-split the
|
||||
trading actor's client is never warmed. Fix: lazily
|
||||
qualify + cache the contract per order request in
|
||||
`handle_order_requests()` by running the same
|
||||
`get_mkt_info(fqme, proxy=...)` ep the feed side uses,
|
||||
plus per-order error relay (`BrokerdError`) so one bad
|
||||
submission can't crash the whole trades dialog (the
|
||||
`TrioTaskExited` storm was teardown cascade from the
|
||||
original raise).
|
||||
|
||||
## Files changed
|
||||
|
||||
- `piker/brokers/ib/broker.py` — thread `proxies` into
|
||||
`handle_order_requests()`; lazy contract qualify on
|
||||
cache-miss; guard `submit_limit()` w/ `BrokerdError`
|
||||
relay; uncomment the (anticipatory) `get_mkt_info`
|
||||
import
|
||||
- `piker/brokers/ib/api.py` — fix the non-f-string
|
||||
raise msg + document the new qualification contract
|
||||
|
||||
## Human edits
|
||||
|
||||
None — committed as generated. Live `ib` order retest
|
||||
performed by the human post-commit.
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
---
|
||||
model: claude-fable-5[1m]
|
||||
service: claude
|
||||
timestamp: 2026-06-10T21:35:49Z
|
||||
git_ref: datad_service
|
||||
diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.md
|
||||
---
|
||||
|
||||
NOTE: diff-ref mode entry (code committed in the same
|
||||
commit as this log); recorded from the live debug
|
||||
session per the `/prompt-io` skill rules.
|
||||
|
||||
> `git log -1 -p --follow -- piker/brokers/ib/broker.py`
|
||||
> `git log -1 -p --follow -- piker/brokers/ib/api.py`
|
||||
|
||||
Key diagnostic chain (from session):
|
||||
|
||||
- pdb showed `Client._contracts == {}` inside
|
||||
`brokerd.ib`'s `submit_limit()`; the cache has
|
||||
exactly TWO write sites: `Client.find_contracts()`
|
||||
(api.py, keys `f'{sym}.{exch}.ib'`) and
|
||||
`symbols.get_mkt_info()` (key `mkt.bs_fqme`, eg.
|
||||
'nvda.nasdaq' — NO `.ib` suffix).
|
||||
- `BrokerdOrder.symbol` arrives as the bs_fqme form
|
||||
('nvda.nasdaq') so ONLY the `get_mkt_info()` write
|
||||
site produces the key `submit_limit()` reads —
|
||||
ie. pre-split it was the feed's in-proc
|
||||
`get_mkt_info(sym, proxy=proxy)` call keeping orders
|
||||
working, NOT `find_contracts()`.
|
||||
- the existing TODO at `symbols.py:642-644` literally
|
||||
predicted this: "this is going to be problematic
|
||||
if/when we split out the datad vs. brokerd actors
|
||||
since the mktmap lookup table will now be
|
||||
inaccessible.."
|
||||
- instance identity verified: `proxy._aio_ns` IS the
|
||||
same `Client` obj as `_accounts2clients[account]`
|
||||
(both sourced from the `load_aio_clients()` cache via
|
||||
`open_client_proxies()`), so a brokerd-side
|
||||
`get_mkt_info(fqme, proxy=proxies[account])` warms
|
||||
exactly the dict `submit_limit()` reads. It also
|
||||
populates `client._cons2mkts` which the
|
||||
position-audit path (`broker.py` backup-table code)
|
||||
needs in this actor anyway.
|
||||
- the `TrioTaskExited` storm in the user's log
|
||||
(`recv_trade_updates`, `open_aio_client_method_relay`
|
||||
aio tasks) is teardown cascade: the raise crashed
|
||||
`handle_order_requests` -> nursery teardown ripped
|
||||
the trio sides of still-running aio relay tasks.
|
||||
Hence the added per-order try/except ->
|
||||
`BrokerdError` relay hardening so a single bad
|
||||
submission degrades to an EMS error msg instead of
|
||||
killing the backend's entire order-ctl dialog.
|
||||
|
||||
Verification: `tests/test_services.py` (5 passed) +
|
||||
`tests/test_ems.py` (6 passed) regression-green; live
|
||||
`ib` submission retest delegated to the human (needs a
|
||||
running TWS/gw).
|
||||
|
|
@ -967,26 +967,8 @@ class Client:
|
|||
|
||||
else:
|
||||
raise
|
||||
# XXX: ambiguous/unqualifiable contracts are
|
||||
# returned as `None` entries by
|
||||
# `qualifyContractsAsync()` (which also logs an
|
||||
# "Ambiguous contract" warning listing the
|
||||
# possible matches) so filter them BEFORE any
|
||||
# attr access B)
|
||||
contracts: list[Contract] = [
|
||||
tract
|
||||
for tract in contracts
|
||||
if tract is not None
|
||||
]
|
||||
if not contracts:
|
||||
raise ValueError(
|
||||
f'No (unambiguous) contract could be '
|
||||
f'qualified for {con!r}\n'
|
||||
f'\n'
|
||||
f'If a stonk, you likely need a (more) '
|
||||
f'venue-qualified fqme,\n'
|
||||
f"eg. 'gld.arca.ib' instead of 'gld.ib'\n"
|
||||
)
|
||||
raise ValueError(f"No contract could be found {con}")
|
||||
|
||||
# pack all contracts into cache
|
||||
for tract in contracts:
|
||||
|
|
@ -1164,16 +1146,10 @@ class Client:
|
|||
try:
|
||||
con: Contract = self._contracts[symbol]
|
||||
except KeyError:
|
||||
# require that the mkt's contract has been previously
|
||||
# qualified and cached (see `.symbols.get_mkt_info()`
|
||||
# which is run for any feed-init OR lazily by the
|
||||
# order-request handler in `.broker`) - ensures we
|
||||
# aren't making orders against unknown contracts.
|
||||
raise RuntimeError(
|
||||
f'Can not order {symbol!r}, '
|
||||
f'no qualified contract cached!?\n'
|
||||
f'_contracts: {list(self._contracts)!r}\n'
|
||||
)
|
||||
# require that the symbol has been previously cached by
|
||||
# a data feed request - ensure we aren't making orders
|
||||
# against non-known prices.
|
||||
raise RuntimeError("Can not order {symbol}, no live feed?")
|
||||
|
||||
try:
|
||||
trade = self.ib.placeOrder(
|
||||
|
|
@ -1632,17 +1608,6 @@ class MethodProxy:
|
|||
self.event_table = event_table
|
||||
self._aio_ns = asyncio_ns
|
||||
|
||||
# request-id counter for correlating each method
|
||||
# call to its (eventual) response; necessary since
|
||||
# the (single, shared) chan has NO other ordering
|
||||
# guarantee whenever a caller task is CANCELLED
|
||||
# (eg. by a search-req timeout) after sending its
|
||||
# request but before consuming the response: the
|
||||
# "orphaned" response otherwise gets mis-delivered
|
||||
# to the next caller causing an off-by-one skew of
|
||||
# every result thereafter!
|
||||
self._mids = itertools.count()
|
||||
|
||||
async def _run_method(
|
||||
self,
|
||||
*,
|
||||
|
|
@ -1656,10 +1621,10 @@ class MethodProxy:
|
|||
|
||||
'''
|
||||
chan = self.chan
|
||||
mid: int = next(self._mids)
|
||||
await chan.send((meth, kwargs, mid))
|
||||
await chan.send((meth, kwargs))
|
||||
|
||||
while not chan.closed():
|
||||
# send through method + ``kwargs: dict`` as pair
|
||||
msg = await chan.receive()
|
||||
|
||||
# TODO: implement reconnect functionality like
|
||||
|
|
@ -1669,51 +1634,24 @@ class MethodProxy:
|
|||
# except ConnectionError:
|
||||
# self.reset()
|
||||
|
||||
match msg:
|
||||
# OUR method-call response B)
|
||||
case {'mid': resp_mid, 'result': res} if (
|
||||
resp_mid == mid
|
||||
):
|
||||
return res
|
||||
# print(f'NEXT MSG: {msg}')
|
||||
|
||||
case {'mid': resp_mid, 'exception': err} if (
|
||||
resp_mid == mid
|
||||
):
|
||||
raise err
|
||||
# TODO: py3.10 ``match:`` syntax B)
|
||||
if 'result' in msg:
|
||||
res = msg.get('result')
|
||||
return res
|
||||
|
||||
# an "orphaned" response to some prior
|
||||
# (cancelled) caller's request; drop it and
|
||||
# keep waiting for ours.
|
||||
case {'mid': resp_mid}:
|
||||
log.warning(
|
||||
f'Dropping stale method-resp,\n'
|
||||
f'mid: {resp_mid} (ours: {mid})\n'
|
||||
f'(a prior caller prolly got '
|
||||
f'cancelled before its resp?)\n'
|
||||
)
|
||||
continue
|
||||
elif 'exception' in msg:
|
||||
err = msg.get('exception')
|
||||
raise err
|
||||
|
||||
# out-of-band (inline) client error: raise
|
||||
# to the current caller as before.
|
||||
case {'exception': err}:
|
||||
raise err
|
||||
elif 'error' in msg:
|
||||
etype, emsg = msg
|
||||
log.warning(f'IB error relay: {emsg}')
|
||||
continue
|
||||
|
||||
case ('error', emsg):
|
||||
log.warning(f'IB error relay: {emsg}')
|
||||
continue
|
||||
|
||||
# routine (api-farm conn) status events
|
||||
# relayed inline by `Client.inline_errors()`;
|
||||
# not a response to our method call so just
|
||||
# log at info and keep waiting.
|
||||
case ('event', emsg):
|
||||
log.info(
|
||||
f'IB status event relay: {emsg}'
|
||||
)
|
||||
continue
|
||||
|
||||
case _:
|
||||
log.warning(f'UNKNOWN IB MSG: {msg}')
|
||||
else:
|
||||
log.warning(f'UNKNOWN IB MSG: {msg}')
|
||||
|
||||
def status_event(
|
||||
self,
|
||||
|
|
@ -1766,31 +1704,22 @@ async def open_aio_client_method_relay(
|
|||
log.info('asyncio `Client` method-proxy SHUTDOWN!')
|
||||
break
|
||||
|
||||
case (meth_name, kwargs, mid):
|
||||
case (meth_name, kwargs):
|
||||
meth_name, kwargs = msg
|
||||
meth = getattr(client, meth_name)
|
||||
|
||||
try:
|
||||
resp = await meth(**kwargs)
|
||||
# echo the msg back tagged with the
|
||||
# req-id for caller-side correlation.
|
||||
chan.send_nowait({
|
||||
'mid': mid,
|
||||
'result': resp,
|
||||
})
|
||||
# echo the msg back
|
||||
chan.send_nowait({'result': resp})
|
||||
|
||||
# XXX: relay ALL (non-cancel) errors to the
|
||||
# `trio`-side caller (which re-raises in the
|
||||
# `MethodProxy._run_method()` frame) instead
|
||||
# of crashing this relay task and thus the
|
||||
# whole proxy/dialog; critical post the
|
||||
# (datad|brokerd)-split where eg. contract
|
||||
# qualification failures are expected to be
|
||||
# caught per-request by order/warmup code!
|
||||
except Exception as err:
|
||||
chan.send_nowait({
|
||||
'mid': mid,
|
||||
'exception': err,
|
||||
})
|
||||
except (
|
||||
RequestError,
|
||||
|
||||
# TODO: relay all errors to trio?
|
||||
# BaseException,
|
||||
) as err:
|
||||
chan.send_nowait({'exception': err})
|
||||
|
||||
case {'error': content}:
|
||||
chan.send_nowait({'exception': content})
|
||||
|
|
|
|||
|
|
@ -89,9 +89,8 @@ from .api import (
|
|||
MethodProxy,
|
||||
)
|
||||
from .symbols import (
|
||||
cache_contract,
|
||||
con2fqme,
|
||||
get_mkt_info,
|
||||
# get_mkt_info,
|
||||
)
|
||||
from .ledger import (
|
||||
norm_trade_records,
|
||||
|
|
@ -139,7 +138,6 @@ async def handle_order_requests(
|
|||
ems_order_stream: tractor.MsgStream,
|
||||
accounts_def: dict[str, str],
|
||||
flows: OrderDialogs,
|
||||
proxies: dict[str, MethodProxy],
|
||||
|
||||
) -> None:
|
||||
|
||||
|
|
@ -182,58 +180,6 @@ async def handle_order_requests(
|
|||
if action in {'buy', 'sell'}:
|
||||
# validate
|
||||
order = BrokerdOrder(**request_msg)
|
||||
fqme: str = order.symbol
|
||||
|
||||
# XXX: lazily qualify and cache the contract for
|
||||
# this mkt since, post the (datad|brokerd)-split,
|
||||
# this trading-only actor will NOT have had its
|
||||
# (api) `Client._contracts` pre-warmed by any
|
||||
# in-proc feed setup (which now runs in the
|
||||
# `datad.ib` sibling); we run the SAME symbology
|
||||
# resolution ep the feed-side uses so the cache
|
||||
# key (`MktPair.bs_fqme`) matches what
|
||||
# `Client.submit_limit()` looks up.
|
||||
# NOTE: normally a no-op since
|
||||
# `open_trade_dialog()` eagerly pre-qualifies all
|
||||
# already-open pp/order mkts at startup; this
|
||||
# only fires for a brand-new (to this daemon)
|
||||
# mkt's first order.
|
||||
if fqme not in client._contracts:
|
||||
proxy: MethodProxy = proxies[account]
|
||||
try:
|
||||
(
|
||||
mkt,
|
||||
details,
|
||||
) = await get_mkt_info(
|
||||
fqme,
|
||||
proxy=proxy,
|
||||
)
|
||||
# XXX: explicit write since the lifo-cache
|
||||
# may deliver a hit (body skipped) when
|
||||
# another acct/client already qualified
|
||||
# this fqme.
|
||||
cache_contract(
|
||||
client,
|
||||
mkt,
|
||||
details.contract,
|
||||
)
|
||||
except Exception as err:
|
||||
log.exception(
|
||||
f'Failed to qualify contract for\n'
|
||||
f'fqme: {fqme!r}\n'
|
||||
)
|
||||
await ems_order_stream.send(
|
||||
BrokerdError(
|
||||
oid=oid,
|
||||
symbol=fqme,
|
||||
reason=(
|
||||
f'No contract could be '
|
||||
f'qualified for {fqme!r}:\n'
|
||||
f'{err!r}'
|
||||
),
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
# XXX: by default 0 tells ``ib_async`` methods that
|
||||
# there is no existing order so ask the client to create
|
||||
|
|
@ -245,34 +191,15 @@ async def handle_order_requests(
|
|||
reqid = int(reqid)
|
||||
|
||||
# call our client api to submit the order
|
||||
# NOTE: guard with order-error relay (vs. crashing
|
||||
# this dialog and thus ALL order ctl for the
|
||||
# backend) so one bad submission can't take down
|
||||
# the daemon's clearing loop.
|
||||
try:
|
||||
reqid = client.submit_limit(
|
||||
oid=order.oid,
|
||||
symbol=fqme,
|
||||
price=order.price,
|
||||
action=order.action,
|
||||
size=order.size,
|
||||
account=acct_number,
|
||||
reqid=reqid,
|
||||
)
|
||||
except Exception as err:
|
||||
log.exception(
|
||||
f'Order submission failed for\n'
|
||||
f'fqme: {fqme!r}\n'
|
||||
)
|
||||
await ems_order_stream.send(
|
||||
BrokerdError(
|
||||
oid=oid,
|
||||
symbol=fqme,
|
||||
reason=f'Submission error: {err!r}',
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
reqid = client.submit_limit(
|
||||
oid=order.oid,
|
||||
symbol=order.symbol,
|
||||
price=order.price,
|
||||
action=order.action,
|
||||
size=order.size,
|
||||
account=acct_number,
|
||||
reqid=reqid,
|
||||
)
|
||||
str_reqid: str = str(reqid)
|
||||
if reqid is None:
|
||||
err_msg = BrokerdError(
|
||||
|
|
@ -621,14 +548,10 @@ async def open_trade_dialog(
|
|||
|
||||
) -> AsyncIterator[dict[str, Any]]:
|
||||
|
||||
# XXX: ONLY adjust the level of this (sub)mod's logger;
|
||||
# attaching a (stderr) handler (via `get_console_log()`)
|
||||
# here would DOUBLE-print every record since the daemon
|
||||
# fixture (`.._daemon._setup_persistent_brokerd()`)
|
||||
# already enables the console handler on the parent
|
||||
# subsys logger which all our records propagate to!
|
||||
if loglevel:
|
||||
log.setLevel(loglevel.upper())
|
||||
get_console_log(
|
||||
level=loglevel,
|
||||
name=__name__,
|
||||
)
|
||||
|
||||
# task local msg dialog tracking
|
||||
flows = OrderDialogs()
|
||||
|
|
@ -865,64 +788,6 @@ async def open_trade_dialog(
|
|||
for msg in order_msgs:
|
||||
await ems_stream.send(msg)
|
||||
|
||||
# XXX: eagerly pre-qualify (and cache) the
|
||||
# contracts for all already-open pps and
|
||||
# orders so that (live) order submission
|
||||
# NEVER pays a first-request qualification
|
||||
# delay; any brand-new mkt is still lazily
|
||||
# qualified by `handle_order_requests()` on
|
||||
# its first submission. NOTE: since this
|
||||
# runs BEFORE the order handler task is
|
||||
# even started, no order can clear until
|
||||
# the warmup completes (early reqs just
|
||||
# buffer in the ems IPC stream) B)
|
||||
warmup_fqmes: set[str] = {
|
||||
msg.symbol
|
||||
for msg in all_positions
|
||||
}
|
||||
warmup_fqmes.update(
|
||||
msg.req.symbol
|
||||
for msg in order_msgs
|
||||
)
|
||||
unique_clients: set[Client] = set(
|
||||
aioclients.values()
|
||||
)
|
||||
if (
|
||||
warmup_fqmes
|
||||
and
|
||||
proxies
|
||||
):
|
||||
a_proxy: MethodProxy = next(
|
||||
iter(proxies.values())
|
||||
)
|
||||
for fqme in warmup_fqmes:
|
||||
try:
|
||||
(
|
||||
mkt,
|
||||
details,
|
||||
) = await get_mkt_info(
|
||||
fqme,
|
||||
proxy=a_proxy,
|
||||
)
|
||||
except Exception as err:
|
||||
# XXX: non-fatal; an
|
||||
# un-warmed mkt just falls
|
||||
# back to the lazy qualify
|
||||
# in the order handler.
|
||||
log.warning(
|
||||
f'Failed to pre-qualify\n'
|
||||
f'fqme: {fqme!r}\n'
|
||||
f'err: {err!r}\n'
|
||||
)
|
||||
continue
|
||||
|
||||
for _client in unique_clients:
|
||||
cache_contract(
|
||||
_client,
|
||||
mkt,
|
||||
details.contract,
|
||||
)
|
||||
|
||||
for client in set(aioclients.values()):
|
||||
trade_event_stream: LinkedTaskChannel = await tn.start(
|
||||
open_trade_event_stream,
|
||||
|
|
@ -936,7 +801,6 @@ async def open_trade_dialog(
|
|||
ems_stream,
|
||||
accounts_def,
|
||||
flows,
|
||||
proxies,
|
||||
)
|
||||
|
||||
# allocate event relay tasks for each client connection
|
||||
|
|
|
|||
|
|
@ -639,40 +639,12 @@ async def get_mkt_info(
|
|||
|
||||
# if possible register the bs_mktid to the just-built
|
||||
# mkt so that it can be retreived by order mode tasks later.
|
||||
# TODO NOTE: this is going to be problematic if/when we split
|
||||
# out the datatd vs. brokerd actors since the mktmap lookup
|
||||
# table will now be inaccessible..
|
||||
if proxy is not None:
|
||||
cache_contract(
|
||||
proxy._aio_ns,
|
||||
mkt,
|
||||
con,
|
||||
)
|
||||
client: Client = proxy._aio_ns
|
||||
client._contracts[mkt.bs_fqme] = con
|
||||
client._cons2mkts[con] = mkt
|
||||
|
||||
return mkt, details
|
||||
|
||||
|
||||
def cache_contract(
|
||||
client: Client,
|
||||
mkt: MktPair,
|
||||
con: ibis.Contract,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Register a (qualified) contract + mkt-info pair on the
|
||||
given (api) `Client`'s actor-local lookup tables.
|
||||
|
||||
Cached under BOTH fqme key-forms since consumers vary:
|
||||
- `mkt.bs_fqme` (eg. 'nvda.nasdaq'): read by
|
||||
`Client.submit_limit()` for order requests,
|
||||
- `mkt.fqme` (eg. 'nvda.nasdaq.ib'): read by the
|
||||
fill-time pp-update (symcache-backup-table) path in
|
||||
`.broker.emit_pp_update()`.
|
||||
|
||||
NOTE: post the (datad|brokerd)-actor-split this MUST be
|
||||
run (in the trading actor) for every mkt either eagerly
|
||||
at `.broker.open_trade_dialog()` startup or lazily per
|
||||
order request; there is no in-proc feed setup doing it
|
||||
implicitly anymore!
|
||||
|
||||
'''
|
||||
client._contracts[mkt.bs_fqme] = con
|
||||
client._contracts[mkt.fqme] = con
|
||||
client._cons2mkts[con] = mkt
|
||||
|
|
|
|||
|
|
@ -93,18 +93,6 @@ async def _setup_persistent_datad(
|
|||
)
|
||||
assert log.name == _util.subsys
|
||||
|
||||
# XXX: ALSO enable console logging for the provider
|
||||
# backend's mod subtree (eg. `piker.brokers.ib.*`)
|
||||
# since (pre the datad|brokerd-split) that was done
|
||||
# by `brokerd`'s fixture; without this all backend
|
||||
# records here fall through to the stdlib's bare
|
||||
# (non-colorized) `logging.lastResort` handler!
|
||||
get_console_log(
|
||||
level=loglevel or tll,
|
||||
name=f'piker.brokers.{brokername}',
|
||||
with_tractor_log=bool(tll),
|
||||
)
|
||||
|
||||
from piker.data import feed
|
||||
assert not feed._bus
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue