From 1da9b760a3bc6b96501110cab2b5d09846d42a1d Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 23 Sep 2026 14:35:21 -0400 Subject: [PATCH] Preserve IB history request bounds on no data Complete HMDS no-data replies without silently retrying a day older. Let `open_history_client()` classify the original request and leave frame stepping to `start_backfill()`. Distinguish `request_end_dt` from `oldest_shm_dt` so the first frame after an empty interval retains its unwritten end bar. Inline the storage time cutoff and drop the unused two-direction `diff_history()` helper. Explain both dates and filtering rules. Add request-boundary tests and real SHM/NativeDB regressions: all 2,004 supplied bars survive, while the closure stays empty. Prompt-IO: ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md (this patch was generated in some part by `codex` using `gpt-6` (`openai`)) --- .../20260923T164656Z_efa72980_prompt_io.md | 64 +++++++ ...20260923T164656Z_efa72980_prompt_io.raw.md | 87 ++++++++++ piker/brokers/ib/feed.py | 46 +++-- piker/tsp/_history.py | 103 ++++++------ tests/test_history_backfill.py | 157 ++++++++++++++++++ tests/test_ib_history.py | 69 ++++++++ 6 files changed, 448 insertions(+), 78 deletions(-) create mode 100644 ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md create mode 100644 ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.raw.md diff --git a/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md b/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md new file mode 100644 index 00000000..ebc1ccd8 --- /dev/null +++ b/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.md @@ -0,0 +1,64 @@ +--- +model: gpt-6 (exact variant not exposed) +service: codex +timestamp: 2026-09-23T16:46:56.830897+00:00 +git_ref: efa72980 +session: 01a0a5b9-f872-7ca3-ab8a-d1cffc14e895 +scope: code +substantive: true +raw_file: 20260923T164656Z_efa72980_prompt_io.raw.md +--- + +## Prompt + +Investigate existing branch fixes before repairing IB day-skipping. +Create an isolated worktree on ib_connectivity_harden and add a test +that checks actual persisted data, leaving live gap repair for later. + +## Response summary + +Preserve IB request boundaries on HMDS no-data responses. Retain the +existing earliest-history classification and bounded reset handling. +Separate requested and published boundaries during backfill so the +first response after an empty interval does not lose its end bar. +Prove supplied bars survive both SHM publication and a fresh parquet +load while a simulated venue closure remains empty. + +## Files changed + +- `piker/brokers/ib/feed.py`: complete HMDS no-data requests unchanged. +- `piker/tsp/_history.py`: filter overlap against published history. +- `tests/test_ib_history.py`: six empty-response boundary cases. +- `tests/test_history_backfill.py`: two SHM/parquet integration cases. + +## Human edits + +The human identified the live gap, required prior-branch discovery, +selected ib_connectivity_harden as the base, and requested stronger +end-to-end evidence. They corrected overstated reproduction claims, +insisted on plain explanations, and pointed out that valid venue +closures must remain gaps. Existing data repair was explicitly deferred. +No direct human source edits have been observed; changes await review. + +## Review follow-up + +The human's tuicr review asked whether None was part of the backfill +protocol and requested plain explanations of the two dates and overlap +filter. Clarified the IB-internal empty-response sentinel and its +translation into NoData/DataUnavailable before reaching the backfiller. +Renamed last_start_dt to request_end_dt and published_start_dt to +oldest_shm_dt. Added a concrete empty-interval example at the filter and +explained that only successful SHM writes move oldest_shm_dt. +These changes clarify naming and documentation without changing the +previously tested request or publication behavior. + +## History-filter review follow-up + +The human flagged the unused append_until_dt argument and confusing +append/prepend terminology in diff_history(). The helper had one +caller, using only prepend_until_dt. Removed the helper and inlined +its inclusive lower-time cutoff beside the SHM duplicate filter. +The human's nearby TODO comment is resolved by this removal; the +replacement comment describes the retained timestamps directly. +The storage comparison remains >= backfill_until_dt; the SHM +comparison remains < oldest_shm_dt. diff --git a/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.raw.md b/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.raw.md new file mode 100644 index 00000000..3958f74f --- /dev/null +++ b/ai/prompt-io/codex/20260923T164656Z_efa72980_prompt_io.raw.md @@ -0,0 +1,87 @@ +--- +model: gpt-6 (exact variant not exposed) +service: codex +timestamp: 2026-09-23T16:46:56.830897+00:00 +git_ref: efa72980 +diff_cmd: git diff efa72980 -- piker/brokers/ib/feed.py piker/tsp/_history.py tests/test_ib_history.py tests/test_history_backfill.py +--- + +## Prompt + +The human requested the IB request-boundary repair in a new worktree +based on ib_connectivity_harden and a test exercising real persistence. +The existing MNQ parquet gap is outside this repair scope. + +## Code output + +> `git diff efa72980 -- piker/brokers/ib/feed.py` + +> `git diff efa72980 -- piker/tsp/_history.py` + +> `git diff efa72980 -- tests/test_ib_history.py` + +> `git diff efa72980 -- tests/test_history_backfill.py` + +get_bars() now completes HMDS no-data responses without changing the +requested date. open_history_client() retains its NoData and earliest +history classification. start_backfill() tracks the published boundary +separately from the request cursor after empty intervals. + +## Unedited progress output + +The new test failed on the original code as intended: the +temporary parquet contained only 4 bars instead of 2,004. +The day-skip caused 2,000 available bars to be omitted. + +The test also includes a simulated venue closure that must remain +a gap. It checks that we fetch available bars without inventing +data for an empty interval. + +The IB fix now preserves all 2,004 bars in parquet. The SHM +assertion exposed a second bug: after stepping over an empty +interval, the backfiller drops the last bar of the next response +as though it were already published. + +I’ll separate the request date from the published-data boundary +so the same test can verify that both memory and disk retain +every available bar. + +## Validation + +Using the existing fsp_backfill_sync/py313 interpreter, with cwd in +ib_history_request_bounds and the Piker import checked against that cwd: + +- tests/test_history_backfill.py: 11 passed in 1.92s. +- tests/test_ib_history.py: 17 passed in 0.36s. +- Ruff on all four changed Python files: all checks passed. +- git diff --check: passed. + +The integration test uses simulated IB bars responses, real IB history +adapter and backfill logic, real SHM, and temporary NativeDB parquet. +It checks a fresh disk load. It does not test a live gateway, actor RPC, +or Qt rendering, and does not prove the origin of the user's live gap. + +## Unedited review response + +`None` is internal to the IB adapter. The backfiller receives +`NoData`, or `DataUnavailable` when the requested date has reached +the known beginning of history. I’ll make that distinction explicit. + +The two dates answer different questions: “where should the next +history query end?” and “what is the oldest bar already in SHM?” +An empty reply changes only the first. I’ll rename them to +`request_end_dt` and `oldest_shm_dt` and explain the filtering with +a concrete example. + +> `git diff efa72980 -- piker/brokers/ib/feed.py piker/tsp/_history.py` + +## History-filter review follow-up + +The human flagged the unused append_until_dt argument and confusing +append/prepend terminology in diff_history(). The helper had one +caller, using only prepend_until_dt. Removed the helper and inlined +its inclusive lower-time cutoff beside the SHM duplicate filter. +The human's nearby TODO comment is resolved by this removal; the +replacement comment describes the retained timestamps directly. +The storage comparison remains >= backfill_until_dt; the SHM +comparison remains < oldest_shm_dt. 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/piker/tsp/_history.py b/piker/tsp/_history.py index 06ef76eb..d0d6bd5f 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -145,29 +145,6 @@ _rt_buffer_start = int((_days_worth - 1) * _secs_in_day) _notify_timeout_s: float = 1 -def diff_history( - array: np.ndarray, - append_until_dt: datetime|None = None, - prepend_until_dt: datetime|None = None, - -) -> np.ndarray: - - # no diffing with tsdb dt index possible.. - if ( - prepend_until_dt is None - and - append_until_dt is None - ): - return array - - times = array['time'] - - if append_until_dt: - return array[times < append_until_dt.timestamp()] - else: - return array[times >= prepend_until_dt.timestamp()] - - def mk_storage_key(mkt: MktPair) -> str: ''' Build the historical storage key for a market. @@ -586,16 +563,24 @@ async def start_backfill( # - after this loop continue to check for other gaps in the # (tsdb) history and (at least report) maybe fill them # from new frame queries to the backend? - last_start_dt: datetime = backfill_from_dt + # start_backfill() asks get_hist() for bars ending at + # request_end_dt. On NoData we move that date backward + # and try an earlier interval, even though no bars arrived. + request_end_dt: datetime = backfill_from_dt + + # oldest_shm_dt is the earliest bar already written to SHM + # by this backfill. It changes only after a successful push. + # NoData changes request_end_dt but leaves this date alone. + oldest_shm_dt: datetime = backfill_from_dt next_prepend_index: int = backfill_from_shm_index est = timezone('EST') - while last_start_dt > backfill_until_dt: + while request_end_dt > backfill_until_dt: log.info( f'Requesting {timeframe}s frame:\n' f'backfill_until_dt: {backfill_until_dt}\n' - f'last_start_dt: {last_start_dt}\n' + f'request_end_dt: {request_end_dt}\n' ) try: ( @@ -604,22 +589,23 @@ async def start_backfill( next_end_dt, ) = await get_hist( timeframe, - end_dt=(end_dt_param := last_start_dt), + end_dt=(end_dt_param := request_end_dt), ) except NoData as nodata: _nodata = nodata - orig_last_start_dt: datetime = last_start_dt + previous_request_end_dt: datetime = request_end_dt gap_report: str = ( - f'EMPTY FRAME for `end_dt: {last_start_dt}`?\n' + f'EMPTY FRAME for `end_dt: {request_end_dt}`?\n' f'{mod.name} -> tf@fqme: {timeframe}@{mkt.fqme}\n' - f'last_start_dt: {orig_last_start_dt}\n\n' + f'request_end_dt: {previous_request_end_dt}\n' + f'\n' f'bf_until: {backfill_until_dt}\n' ) # EMPTY FRAME signal with 3 (likely) causes: # # 1. range contains legit gap in venue history # 2. history actually (edge case) **began** at the - # value `last_start_dt` + # value `request_end_dt` # 3. some other unknown error (ib blocking the # history-query bc they don't want you seeing how # they cucked all the tinas.. like with options @@ -630,13 +616,13 @@ async def start_backfill( # as maybe indicated by the backend to see if we # can get older data before this possible # "history gap". - last_start_dt: datetime = last_start_dt.subtract( + request_end_dt = request_end_dt.subtract( seconds=def_frame_duration.total_seconds() ) gap_report += ( f'Decrementing `end_dt` and retrying with,\n' f'def_frame_duration: {def_frame_duration}\n' - f'(new) last_start_dt: {last_start_dt}\n' + f'(new) request_end_dt: {request_end_dt}\n' ) log.warning(gap_report) # skip writing to shm/tsdb and try the next @@ -657,7 +643,7 @@ async def start_backfill( f'fqme: {mkt.fqme}\n' f'timeframe: {timeframe}\n' - f'last_start_dt: {last_start_dt}\n' + f'request_end_dt: {request_end_dt}\n' f'bf_until: {backfill_until_dt}\n' ) # UGH: what's a better way? @@ -712,7 +698,7 @@ async def start_backfill( # await tractor.pause() expected_dur: Interval = ( - last_start_dt.subtract( + request_end_dt.subtract( seconds=timeframe # ^XXX, always "up to" the bar *before* ) @@ -747,7 +733,8 @@ async def start_backfill( log.warning( f'{timeframe}s-series {reason} detected!\n' f'fqme: {mkt.fqme}\n' - f'last_start_dt: {last_start_dt}\n\n' + f'request_end_dt: {request_end_dt}\n' + f'\n' f'recv interval: {recv_frame_dur}\n' f'expected interval: {expected_dur}\n\n' @@ -756,26 +743,33 @@ async def start_backfill( ) # await tractor.pause() - to_store: np.ndarray = diff_history( - array, - prepend_until_dt=backfill_until_dt, - ) - # The requested end bar is already the oldest published - # sample. Keep it for storage merge, but do not - # consume another physical SHM row for the overlap. + # Stop retrieving history at backfill_until_dt. Keep + # that timestamp for the storage merge, plus newer bars; + # discard any older bars included in this provider frame. + to_store: np.ndarray = array[ + array['time'] >= backfill_until_dt.timestamp() + ] + # Keep only bars older than SHM's earliest written bar. + # A bar with the same timestamp already exists in SHM; + # writing it again would duplicate that bar. + # + # Example: SHM starts at 12:00. An empty response moves + # request_end_dt to 11:26:40, but SHM still starts at + # 12:00. If the next reply includes an 11:26:40 bar, we + # must keep it: asking for that time did not write it. to_push: np.ndarray = to_store[ - to_store['time'] < last_start_dt.timestamp() + to_store['time'] < oldest_shm_dt.timestamp() ] ln: int = len(to_push) if ln: log.info( - f'{ln} bars for {next_start_dt} -> {last_start_dt}' + f'{ln} bars for {next_start_dt} -> {request_end_dt}' ) else: log.warning( f'0 BARS TO PUSH after diff!?\n' - f'{next_start_dt} -> {last_start_dt}' + f'{next_start_dt} -> {request_end_dt}' f'\n' f'This might mean we rxed a gap frame which starts BEFORE,\n' f'backfill_until_dt: {backfill_until_dt}\n' @@ -812,7 +806,7 @@ async def start_backfill( ): log.info( f'Writing {len(to_store)} frame to storage:\n' - f'{next_start_dt} -> {last_start_dt}' + f'{next_start_dt} -> {request_end_dt}' ) await storage.update_ohlcv( mk_storage_key(mkt), @@ -835,7 +829,10 @@ async def start_backfill( # decrement next prepend point next_prepend_index = next_prepend_index - ln - last_start_dt = next_start_dt + request_end_dt = next_start_dt + # The push succeeded: these bars now exist in SHM. + # Use their start to exclude duplicates next time. + oldest_shm_dt = next_start_dt # Stop if we've hit buffer start if next_prepend_index <= 0: @@ -848,7 +845,8 @@ async def start_backfill( except ValueError as ve: _ve = ve log.error( - f'Shm prepend OVERRUN on: {next_start_dt} -> {last_start_dt}?' + f'Shm prepend OVERRUN on: {next_start_dt} -> ' + f'{request_end_dt}?' ) if next_prepend_index < ln: @@ -867,12 +865,15 @@ async def start_backfill( update_start_on_prepend=update_start_on_prepend, ) next_prepend_index = next_prepend_index - ln - last_start_dt = next_start_dt + request_end_dt = next_start_dt + # The push succeeded: these bars now exist in SHM. + # Use their start to exclude duplicates next time. + oldest_shm_dt = next_start_dt shm_full = True log.info( f'Shm pushed {ln} frame:\n' - f'{next_start_dt} -> {last_start_dt}' + f'{next_start_dt} -> {request_end_dt}' ) await notify_backfill( diff --git a/tests/test_history_backfill.py b/tests/test_history_backfill.py index 2255f585..d0b19ead 100644 --- a/tests/test_history_backfill.py +++ b/tests/test_history_backfill.py @@ -706,3 +706,160 @@ def test_null_repair_falls_back_without_debug_pause( frame[field][-1], ] assert frame['volume'].tolist() == [5, 0, 0, 0, 6] + + +@pytest.mark.parametrize('empty_reply', ['hmds_error', 'empty_bars']) +def test_ib_empty_interval_preserves_older_bars_in_parquet( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + empty_reply: str, +) -> None: + ''' + An empty IB interval must not skip a day of available history. + + Run the real IB adapter, startup backfiller, SHM publication, + and NativeDB writes against a simulated IB bars endpoint. A + known empty interval separates recent bars from older trades. + The old HMDS handler jumped a day back, so those older trades + never reached SHM or parquet. Keep the legitimate empty interval + and assert every supplied trade survives a fresh storage load. + This covers data ingestion and persistence, not Qt rendering. + + ''' + from ib_async import RequestError + from pendulum import duration + from piker.brokers.ib import feed as ib_feed + + epoch: int = 1790065963 + requests: list[float|None] = [] + finished = trio.Event() + mkt = SimpleNamespace( + fqme='mnq.cme.20261218.ib', + dst=SimpleNamespace(atype='future'), + src=SimpleNamespace(atype='fiat'), + get_fqme=lambda **kwargs: 'mnq.cme.20261218.ib', + get_bs_fqme=lambda **kwargs: 'mnq.cme.20261218', + ) + + def bars(offsets) -> np.ndarray: + frame = np.zeros(len(offsets), dtype=_bar_load_dtype) + frame['time'] = np.asarray(offsets) + epoch + for field in ('open', 'high', 'low', 'close'): + frame[field] = 30000 + np.asarray(offsets) / 4 + frame['volume'] = 1 + frame['count'] = 1 + return frame + + class Proxy: + async def maybe_get_head_time(self, **kwargs): + return from_timestamp(epoch - 864000) + + async def bars(self, **kwargs): + end = kwargs['end_dt'] + end_t = None if end is None else end.timestamp() + requests.append(end_t) + if end_t is None: + frame = bars([4001, 4002]) + elif end_t == epoch + 4001: + if empty_reply == 'hmds_error': + raise RequestError( + 1, 162, 'HMDS query returned no data', + ) + frame = bars([]) + elif end_t == epoch + 2001: + frame = bars(list(range(1, 2002))) + else: + # Model the old adapter's day-skipped response. + # It falls before the stored endpoint and is dropped. + assert end_t == epoch + 4001 - 86400 + frame = bars([-82400, -82399]) + native = [ + SimpleNamespace(date=from_timestamp(timestamp)) + for timestamp in frame['time'] + ] + return native, frame, duration(seconds=2000) + + @asynccontextmanager + async def fake_data_client(): + yield Proxy() + + class Sampler: + async def send(self, msg) -> None: + pass + + # Preserve the real null-repair pass; signal when startup ends. + from piker.tsp import _history + repair = _history.maybe_fill_null_segments + + async def finish_repair(**kwargs) -> None: + await repair(**kwargs) + finished.set() + + monkeypatch.setattr( + ib_feed, 'open_data_client', fake_data_client, + ) + monkeypatch.setattr( + _history, 'maybe_fill_null_segments', finish_repair, + ) + expected = bars(list(range(2002)) + [4001, 4002]) + + async def main() -> None: + storage = NativeStorageClient(tmp_path) + await storage.update_ohlcv(mkt.fqme, bars([0, 1]), 1) + key = f'test_ib_request_bounds_{uuid4().hex}' + with trio.fail_after(10): + async with tractor.open_root_actor( + name=key, + tpt_bind_addrs=[('127.0.0.1', 0)], + ): + shm, opened = maybe_open_shm_array( + key=key, + size=5000, + dtype=np.dtype(_ohlc_dtype), + append_start_index=4500, + ) + assert opened + async with trio.open_nursery() as nursery: + nursery.start_soon(partial( + tsdb_backfill, + mod=SimpleNamespace( + name='ib', + open_history_client=( + ib_feed.open_history_client + ), + ), + storemod=SimpleNamespace( + ohlc_key_map=ohlc_key_map, + ), + storage=storage, + mkt=mkt, + shm=shm, + timeframe=1, + sampler_stream=Sampler(), + )) + await finished.wait() + nursery.cancel_scope.cancel() + # Read actual parquet through a new client to avoid + # accidentally validating only NativeDB's cache. + fresh = NativeStorageClient(tmp_path) + loaded, _, _ = await fresh.load(mkt.fqme, 1) + for field in ( + 'time', 'open', 'high', 'low', 'close', 'volume', + ): + np.testing.assert_array_equal( + loaded[field], expected[field], + err_msg=f'parquet field: {field}', + ) + np.testing.assert_array_equal( + shm.array[field], expected[field], + err_msg=f'SHM field: {field}', + ) + assert requests == [ + None, epoch + 4001, epoch + 2001, + ] + # Exactly the simulated closure remains. Do not + # manufacture bars to force timestamp continuity. + gaps = np.diff(loaded['time']) + assert gaps[gaps > 1].tolist() == [2000] + + trio.run(main) diff --git a/tests/test_ib_history.py b/tests/test_ib_history.py index 33acfa86..9600ac0f 100644 --- a/tests/test_ib_history.py +++ b/tests/test_ib_history.py @@ -586,3 +586,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)