From 02bb193f87b4beae5a75f1e6b63e9bd5201c65f2 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 23 Oct 2026 11:35:31 -0400 Subject: [PATCH] .ib.feed: backport no-data request boundaries PARTIAL BACKPORT -- IB adapter and six boundary cases only. The source also fixes history-layer endpoint/filtering behavior; those changes and its SHM/NativeDB regressions are NOT imported. Import the exact `piker/brokers/ib/feed.py` delta. Complete HMDS no-data replies at the requested date instead of silently walking back a day. Wake the waiter and let `open_history_client()` turn the internal empty sentinel into `NoData` or `DataUnavailable`. Deats, - Retain the original parametrized request-boundary test body and its six cases, adding only its required module import. - Add LOCAL real-bridge history recovery coverage after the reset backport: production proxy/relay/reset tasks, replayed external IB/VNC input, and a real farm event before retrying 1s/60s bars. - Leave `request_end_dt` versus `oldest_shm_dt`, storage cutoff filtering, `diff_history()` removal and full-bar retention coverage to the downstream history-layer commit. Downstream replay, Retain those omitted changes even if the adapter hunks vanish. Preserve the local real-bridge regression when reconciling tests. The resulting adapter file matches the source byte-for-byte; that equivalence does not apply to the whole source commit. Partial-backport-of: 1da9b760a3bc6b96501110cab2b5d09846d42a1d Source-author: goodboy Prompt-IO: ai/prompt-io/opencode/20261022T212248Z_dc618229_ib_recovery_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-6-astra` (`openai`)) --- piker/brokers/ib/feed.py | 46 +++++------- tests/test_ib_history.py | 72 +++++++++++++++++- tests/test_ib_quote_restart.py | 131 +++++++++++++++++++++++++++++++++ 3 files changed, 221 insertions(+), 28 deletions(-) diff --git a/piker/brokers/ib/feed.py b/piker/brokers/ib/feed.py index f5ec0bf3..ff5b7576 100644 --- a/piker/brokers/ib/feed.py +++ b/piker/brokers/ib/feed.py @@ -38,7 +38,6 @@ from async_generator import aclosing import ib_async as ibis import numpy as np from pendulum import ( - now, from_timestamp, Duration, duration as mk_duration, @@ -221,7 +220,11 @@ async def open_history_client( f'mean: {mean}' ) - # could be trying to retreive bars over weekend + # get_bars() uses None internally for an empty reply. + # The backfiller never receives that sentinel: translate + # it into NoData, or DataUnavailable at the known start + # of the instrument's history. A venue closure can also + # yield an empty reply without exhausting all history. if out is None: log.error( f"No bars starting at {end_dt!r} !?!?" @@ -456,10 +459,6 @@ async def get_bars( # how long before we trigger a feed reset (seconds) feed_reset_timeout: float = 3, - # how many days to subtract before giving up on further - # history queries for instrument, presuming that most don't - # not trade for a week XD - max_nodatas: int = 6, max_failed_resets: int = 6, task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED, @@ -479,7 +478,6 @@ async def get_bars( `.ib.api.Client` methods. ''' - nodatas_count: int = 0 cancelled_count: int = 0 failed_resets: int = 0 pacing_reset_pending: bool = False @@ -496,8 +494,8 @@ async def get_bars( async def query(): - nonlocal result, data_cs, end_dt - nonlocal nodatas_count, cancelled_count, failed_resets + nonlocal result, data_cs + nonlocal cancelled_count, failed_resets nonlocal pacing_reset_pending, pacing_reset_completed while True: @@ -625,27 +623,21 @@ async def get_bars( # the frame dt index since the upper layer may # be doing so concurrently and we don't want to # be delivering frames that weren't asked for. - # try to decrement start point and look further back - # end_dt = end_dt.subtract(seconds=2000) - logmsg = "SUBTRACTING DAY from DT index" - if end_dt is not None: - end_dt = end_dt.subtract(days=1) - elif end_dt is None: - end_dt = now().subtract(days=1) - log.warning( f'NO DATA found ending @ {end_dt}\n' - + logmsg ) - - if nodatas_count >= max_nodatas: - raise DataUnavailable( - f'Presuming {fqme} has no further history ' - f'after {max_nodatas} tries..' - ) - - nodatas_count += 1 - continue + # Match get_bars()'s empty-bars return: None is + # internal to the IB adapter, not its backfill + # protocol. open_history_client() converts it + # to NoData (or DataUnavailable at history start). + # Wake the waiting get_bars() task before exiting + # so it does not mistake this reply for a timeout. + result = None + failed_resets = 0 + result_ready.set() + if data_cs: + data_cs.cancel() + return None elif ( 'API historical data query cancelled' diff --git a/tests/test_ib_history.py b/tests/test_ib_history.py index a127d726..3df2b4e2 100644 --- a/tests/test_ib_history.py +++ b/tests/test_ib_history.py @@ -1,5 +1,5 @@ ''' -IB history adapter regressions backported from 0846cbd4. +IB history adapter regressions backported from 0846cbd4 and 1da9b760. ''' from types import SimpleNamespace @@ -7,6 +7,7 @@ from types import SimpleNamespace from ib_async import RequestError import numpy as np from pendulum import ( + datetime as pdatetime, duration, ) import pytest @@ -514,3 +515,72 @@ def test_completed_pacing_reset_restarts_timeout_wait( trio.run(main) assert reset_types == ['data', 'connection'] + + +@pytest.mark.parametrize('empty_reply', ['hmds_error', 'empty_bars']) +@pytest.mark.parametrize( + 'boundary', ['latest', 'after_head', 'head'], +) +def test_empty_history_preserves_request_boundary( + monkeypatch: pytest.MonkeyPatch, + empty_reply: str, + boundary: str, +) -> None: + ''' + Empty IB responses retain their original request metadata. + + Exercise open_history_client() and get_bars() together. The + latest request must not fall back to yesterday; a dated request + must not issue another date or become historical exhaustion + unless it has actually reached the known first timestamp. + + ''' + from contextlib import asynccontextmanager + from piker.brokers import NoData + + head = pdatetime(2026, 9, 18, tz='UTC') + end = { + 'latest': None, + 'after_head': head.add(days=4), + 'head': head, + }[boundary] + requests: list = [] + + class Proxy: + async def maybe_get_head_time(self, **kwargs): + return head + + async def bars(self, **kwargs): + requests.append(kwargs['end_dt']) + if empty_reply == 'hmds_error': + raise RequestError( + 1, 162, 'HMDS query returned no data', + ) + return [], np.empty(0), duration(seconds=2000) + + @asynccontextmanager + async def fake_data_client(): + yield Proxy() + + monkeypatch.setattr( + ib_feed, 'open_data_client', fake_data_client, + ) + mkt = SimpleNamespace( + dst=SimpleNamespace(atype='future'), + get_bs_fqme=lambda **kwargs: 'mnq.cme.20261218', + ) + + async def main() -> None: + with trio.fail_after(1): + async with ib_feed.open_history_client(mkt) as pair: + get_hist, _ = pair + expected = ( + DataUnavailable if boundary == 'head' else NoData + ) + with pytest.raises(expected) as exc: + await get_hist(1, end_dt=end) + if expected is NoData: + assert exc.value.info['end_dt'] == end + assert requests == [end] + + trio.run(main) diff --git a/tests/test_ib_quote_restart.py b/tests/test_ib_quote_restart.py index 424a5d98..c076ca7c 100644 --- a/tests/test_ib_quote_restart.py +++ b/tests/test_ib_quote_restart.py @@ -158,3 +158,134 @@ def test_quote_restart_preserves_actor( assert not errors, errors trio.run(main) + + +async def exercise_history_recovery(timeframe: int) -> None: + ''' + Replay pacing and farm recovery across the production IB adapter. + + Keep Client.bars(), MethodProxy, both relay tasks, get_bars(), + wait_on_data_reset() and data_reset_hack() intact. Replace only + external contract discovery, historical replies and VNC input. + Deliver the farm event through IB.errorEvent after the real reset + task is waiting, and reject a retry before that recovery signal. + + This covers the bridge missing from the imported policy tests; + it neither sends VNC keys nor connects to a broker or storage. + + ''' + requests: int = 0 + reset_calls: int = 0 + recovery_sent: bool = False + reset_started = trio.Event() + contract = Contract(symbol='QQQ') + reason = 'HMDS data farm connection is OK:ushmds' + + class ReplayIB(IB): + ''' + Use real IB events with scripted historical responses. + + ''' + async def reqHistoricalDataAsync(self, **kwargs) -> list: + ''' + Reject the first request; return empty data after reset. + + ''' + nonlocal requests + requests += 1 + if requests == 1: + raise RequestError(1, 162, feed._pacing) + assert recovery_sent + assert requests == 2 + return [] + + ib = ReplayIB() + client = Client(ib=ib, config={'vnc_addrs': []}) + + async def find_contracts(fqme: str) -> list[Contract]: + ''' + Return the replay contract without network qualification. + + ''' + return [contract] + + async def vnc_input(client: Client) -> None: + ''' + Record completion of external reset input on asyncio. + + ''' + nonlocal reset_calls + reset_calls += 1 + assert reset_calls == 1 + reset_started.set() + await asyncio.sleep(0) + + async def emit_recovery() -> None: + ''' + Deliver the Gateway status through Client.inline_errors(). + + ''' + nonlocal recovery_sent + recovery_sent = True + ib.errorEvent.emit(-1, 2106, reason, contract) + + with pytest.MonkeyPatch.context() as patch: + patch.setattr(client, 'find_contracts', find_contracts) + patch.setattr(_util, '_reset_tech', 'vnc') + patch.setattr(_util, 'vnc_click_hack', vnc_input) + async with open_client_proxy(client) as proxy: + async def gateway_status() -> None: + ''' + Wait for the actual farm-event consumer to be ready. + + ''' + await reset_started.wait() + event = proxy.event_table[reason] + while not event.statistics().tasks_waiting: + await trio.lowlevel.checkpoint() + await tractor.to_asyncio.run_task(emit_recovery) + + async with trio.open_nursery() as nursery: + nursery.start_soon(gateway_status) + result, reset_used = await feed.get_bars( + proxy, + 'qqq.nasdaq', + timeframe, + ) + assert result is None + assert reset_used + assert not feed._data_reset_lock.locked() + + assert requests == 2 + assert reset_calls == 1 + + +def test_history_pacing_through_real_bridge(reg_addr) -> None: + ''' + Pacing recovery must complete before retrying either bar period. + + Run the production proxy and reset tasks in an infected actor for + both 1s and 60s history. A returned empty frame must release the + request rather than trigger more resets. Repeat on the same actor + to expose leaked ownership from the first request. + + ''' + async def main() -> None: + with trio.fail_after(10): + async with ( + tractor.open_root_actor(registry_addrs=[reg_addr]), + tractor.open_nursery() as nursery, + ): + portal = await nursery.start_actor( + 'ib_history_replay', + enable_modules=[__name__], + infect_asyncio=True, + ) + for timeframe in (1, 60): + await portal.run( + exercise_history_recovery, + timeframe=timeframe, + ) + await portal.cancel_actor() + + trio.run(main)