From eed58566226d5476b548c6dd3bb6a44fcd6325f9 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 24 Sep 2026 20:20:20 -0400 Subject: [PATCH] Retry incomplete IB schedule coverage A one-day MNQ schedule reply can start in the next session, leaving the gap's earlier trading hours uncovered. Widen `Client.history_schedule()` lookbacks up to 31 days, keeping the requested end fixed and sharing one timeout across attempts. Preserve unknown coverage when widening cannot cover the range. Share `SECONDS_PER_DAY` between query and reservation arithmetic, with one explanation of elapsed versus calendar/session days. Replay the captured MNQ replies through the allocator: reserve 51,824 candidate slots while excluding the maintenance closure. Cover bounded retries and timeout cancellation, and document the query contract. This does not repair existing SHM or parquet. Prompt-IO: ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md (this patch was generated in some part by `codex` using `gpt-6` (`openai`)) --- ...5Z_441358e2_schedule_coverage_prompt_io.md | 45 ++++ ...41358e2_schedule_coverage_prompt_io.raw.md | 66 ++++++ docs/history_segment_backfill.rst | 8 +- piker/brokers/ib/api.py | 49 +++-- piker/calc.py | 5 + piker/tsp/_reserve.py | 6 +- tests/providers/test_ib_schedule.py | 192 +++++++++++++++++- 7 files changed, 352 insertions(+), 19 deletions(-) create mode 100644 ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md create mode 100644 ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md diff --git a/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md b/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md new file mode 100644 index 00000000..a4e4f882 --- /dev/null +++ b/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.md @@ -0,0 +1,45 @@ +--- +model: gpt-6 +service: codex +timestamp: 2026-09-24T23:24:05.525363+00:00 +git_ref: 441358e2 +scope: code +substantive: true +raw_file: 20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md +--- + +## Prompt + +User: "do it", approving the proposed bounded wider-window retry +for incomplete schedule coverage and its MNQ regression test. +Previous live qualification showed a one-day request began after +the gap's left edge; a wider query covered it and excluded the +one-hour venue closure. + +## Response summary + +Added bounded backwards widening under one timeout, preserving actual +coverage and fixed end bounds. Added captured-response allocation, +unknown-coverage, and retry-timeout tests, and updated API documentation. +39 selected tests and Ruff passed. Live data was not changed. + +## Files changed + +- piker/brokers/ib/api.py — bounded schedule coverage retries. +- tests/providers/test_ib_schedule.py — captured MNQ and retry regressions. +- docs/history_segment_backfill.rst — retry contract and limits. + +## Human edits + +The user established schedule-driven allocation, explicit unknown +coverage, and avoiding reindexing as the design constraints, supplied +the MNQ gap, and authorized this focused coverage fix. No direct human +source edits occurred during this patch; review is pending. + +## Follow-up review + +The user requested factoring and documenting the repeated 86400 +conversion. Added piker.calc.SECONDS_PER_DAY, documented elapsed +versus calendar/session days once, and reused it in the IB query +and reservation span check. Named the client retry limit locally. +Schedule tests: 11 passed; Ruff passed after this adjustment. diff --git a/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md b/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md new file mode 100644 index 00000000..bf12dc7e --- /dev/null +++ b/ai/prompt-io/codex/20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md @@ -0,0 +1,66 @@ +--- +model: gpt-6 +service: codex +timestamp: 2026-09-24T23:24:05.525363+00:00 +git_ref: 441358e2 +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py` + +> `git diff HEAD~1..HEAD -- tests/providers/test_ib_schedule.py` + +> `git diff HEAD~1..HEAD -- docs/history_segment_backfill.rst` + +Client.history_schedule() widens incomplete left coverage by doubling +numDays up to 31 while preserving endDateTime and actual returned +coverage. All attempts share an asyncio timeout. Missing right coverage +returns unchanged because extending the lookback cannot repair it. +Tests replay the captured MNQ schedules through the real client and +reservation allocator, check bounded incomplete coverage and retry +cancellation under one timeout. Documentation describes these limits. + +## User-facing output + +The captured MNQ case now passes through the real client method +and allocator: the request widens from one day to two, then +reserves 51,824 slots while leaving the venue closure compressed. + +The tests also cover permanently incomplete replies and a timeout +during the retry. I’m running the related regression suites now. + +All 39 selected tests passed, and Ruff passed. The reservation +suite needed a retry outside the sandbox because Trio’s local +wakeup socket was blocked. + +The fix and regression tests are ready for review. I’m recording +the prompt provenance; live SHM and parquet remain unchanged. + +## Validation commands and results + +Worktree-local py313 in the existing Nix environment; import piker +resolved inside this worktree. + +- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/providers/test_ib_schedule.py tests/providers/test_history_reservations.py + Sandbox: 11 passed then PermissionError creating Trio wakeup socket. + Escalated rerun: 22 passed in 1.02s; fake IB, disposable test SHM. +- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/test_ib_history.py + 17 passed in 0.69s. +- ruff check piker/brokers/ib/api.py tests/providers/test_ib_schedule.py + All checks passed (existing deprecated configuration warning). +- git diff --check: passed. + +No live service restart, live history repair, staging, or commit. + +## Follow-up user review and generated changes + +User: "feels like this same magic # from the prior commit should be +factored and commented once? `num_days: int = max(1, ceil(seconds / +86400))` other then that all looks decent" + +> `git diff HEAD~1..HEAD -- piker/calc.py piker/tsp/_reserve.py` + +Added SECONDS_PER_DAY for both consumers, with one comment explaining +fixed elapsed days versus calendar days and trading sessions. Named +the client's 31-day retry limit locally. After this change the +schedule suite passed 11 tests and Ruff passed all four Python files. diff --git a/docs/history_segment_backfill.rst b/docs/history_segment_backfill.rst index 92a535da..7207d45b 100644 --- a/docs/history_segment_backfill.rst +++ b/docs/history_segment_backfill.rst @@ -21,7 +21,13 @@ Schedule queries for UTC epoch bounds. The response includes actual coverage, timezone, regular-hours policy, and half-open trading sessions. This wraps ``Client.history_schedule()`` through the existing asyncio method proxy. -The first pass limits a query to 31 days and times out after 10 seconds. +The first pass limits a query to 31 days. If IB returns coverage +starting after the requested left edge, double the lookback up to +31 days while retaining the requested end timestamp. All attempts +share one 10-second timeout. A missing right edge cannot be repaired +by widening backwards; return the actual partial coverage so the +allocator leaves that gap unresolved. The same applies when the +31-day lookback still leaves the left edge uncovered. Historical queries must match bars' ``useRTH=False`` selection. Do not extrapolate sessions outside response coverage. Empty schedules, diff --git a/piker/brokers/ib/api.py b/piker/brokers/ib/api.py index d9edc2e0..7565e1ca 100644 --- a/piker/brokers/ib/api.py +++ b/piker/brokers/ib/api.py @@ -363,10 +363,13 @@ class Client: Return UTC coverage separately from sessions. Missing coverage is unknown, never evidence that the venue was closed. - Match bars()'s extended-hours policy by default. + Match bars()'s extended-hours policy by default. Retry missing + left coverage with a wider lookback, up to 31 days, sharing + `timeout_s` across schedule requests. Keep the end fixed. ''' from math import ceil + from piker.calc import SECONDS_PER_DAY from .venues import normalize_schedule if start_dt.tzinfo is None or end_dt.tzinfo is None: @@ -374,23 +377,43 @@ class Client: seconds: float = ( end_dt.astimezone(UTC) - start_dt.astimezone(UTC) ).total_seconds() - if not 0 < seconds <= 31 * 86400: + # Bound schedule lookback and retries to one elapsed month. + max_schedule_days: int = 31 + if not 0 < seconds <= max_schedule_days * SECONDS_PER_DAY: raise ValueError('Schedule range must be within 31 days') contract: Contract = (await self.find_contracts(fqme))[0] - schedule = await asyncio.wait_for( - self.ib.reqHistoricalScheduleAsync( - contract, - numDays=max(1, ceil(seconds / 86400)), - endDateTime=end_dt.astimezone(UTC).strftime( - '%Y%m%d-%H:%M:%S' - ), - useRTH=use_rth, - ), - timeout=timeout_s, + num_days: int = max(1, ceil(seconds / SECONDS_PER_DAY)) + end_stamp: str = end_dt.astimezone(UTC).strftime( + '%Y%m%d-%H:%M:%S' ) + # IB can return a single dated session for a one-day query, + # even when the requested interval starts in the preceding + # session. Widen the lookback, keeping the requested end fixed. + # Doubling bounds request count; 31 days caps provider work. + # All attempts share one timeout, rather than renewing it. # TODO: explicit cancellation of the underlying IB request # requires a request-id exposing ib_async schedule wrapper. - return normalize_schedule(schedule, use_rth) + async with asyncio.timeout(timeout_s): + while True: + schedule = await self.ib.reqHistoricalScheduleAsync( + contract, + numDays=num_days, + endDateTime=end_stamp, + useRTH=use_rth, + ) + coverage: dict = normalize_schedule(schedule, use_rth) + if ( + coverage['start'] <= start_dt.timestamp() + or + coverage['end'] < end_dt.timestamp() + or + num_days == max_schedule_days + ): + # Widening backwards cannot fix missing coverage + # at the right edge. Preserve actual coverage for + # the allocator to reject any unresolved interval. + return coverage + num_days = min(max_schedule_days, num_days * 2) async def bars( self, diff --git a/piker/calc.py b/piker/calc.py index d629b28c..442ec411 100644 --- a/piker/calc.py +++ b/piker/calc.py @@ -23,6 +23,11 @@ import itertools from bidict import bidict +# Fixed elapsed-day conversion for timestamp arithmetic. Calendar +# days may span 23 or 25 hours across DST; trading sessions vary too. +SECONDS_PER_DAY: int = 24 * 60 * 60 + + _mag2suffix = bidict({3: 'k', 6: 'M', 9: 'B'}) diff --git a/piker/tsp/_reserve.py b/piker/tsp/_reserve.py index cb11bff2..01bca2ad 100644 --- a/piker/tsp/_reserve.py +++ b/piker/tsp/_reserve.py @@ -25,6 +25,7 @@ import math import numpy as np +from piker.calc import SECONDS_PER_DAY from piker.log import get_logger log = get_logger(__name__) @@ -102,10 +103,9 @@ async def reserve_history_gaps( inserts: dict[int, np.ndarray] = {} ranges: list[tuple[float, float]] = [] budget: int = max_rows - # Match IB Client.history_schedule()'s first-pass 31-day bound; - # 86400 converts elapsed days to seconds, not local venue days. + # Match IB Client.history_schedule()'s first-pass 31-day bound. # TODO: advertise query limits through backend history config. - max_schedule_span_s: int = 31 * 86400 + max_schedule_span_s: int = 31 * SECONDS_PER_DAY for offset in reversed(gaps[-newest_gaps:]): left: float = float(times[offset]) right: float = float(times[offset + 1]) diff --git a/tests/providers/test_ib_schedule.py b/tests/providers/test_ib_schedule.py index 6e7615c6..88553aec 100644 --- a/tests/providers/test_ib_schedule.py +++ b/tests/providers/test_ib_schedule.py @@ -99,11 +99,11 @@ def test_schedule_request_uses_extended_hours_and_utc(): ) out = asyncio.run(Client.history_schedule( client, 'mnq.cme.20261218', - datetime(2026, 9, 23, 0, tzinfo=UTC), + datetime(2026, 9, 23, 4, tzinfo=UTC), datetime(2026, 9, 24, 1, tzinfo=UTC), )) assert calls == [('contract', { - 'numDays': 2, 'endDateTime': '20260924-01:00:00', + 'numDays': 1, 'endDateTime': '20260924-01:00:00', 'useRTH': False, })] assert out['timezone'] == 'America/New_York' @@ -173,3 +173,191 @@ def test_schedule_duration_counts_elapsed_time_across_dst(): datetime(2026, 11, 1, tzinfo=zone), datetime(2026, 11, 2, tzinfo=zone), )) + + +def mnq_schedule(wider: bool = False) -> SimpleNamespace: + ''' + Reproduce the dated US/Central sessions captured from live IB. + + ''' + starts: list[str] = ['20260923-17:00:00'] + ends: list[str] = ['20260924-16:00:00'] + if wider: + starts.insert(0, '20260922-17:00:00') + ends.insert(0, '20260923-16:00:00') + sessions: list[SimpleNamespace] = [] + start: str + end: str + for start, end in zip(starts, ends): + sessions.append(SimpleNamespace( + startDateTime=start, + endDateTime=end, + )) + return SimpleNamespace( + timeZone='US/Central', + startDateTime=starts[0], + endDateTime=ends[-1], + sessions=sessions, + ) + + +def test_schedule_retry_reserves_mnq_gap_outside_closure(): + ''' + One elapsed day returned only MNQ's newest trading session. + + Its coverage began after the left gap margin, so the allocator + could not reserve any slots. Replay both captured IB responses + through Client.history_schedule() and reserve_history_gaps(). + Assert that widening keeps the end fixed, reserves both trading + portions, and leaves the one-hour maintenance closure compressed. + This uses fake transport and arrays, with no live SHM or storage. + + ''' + import numpy as np + from piker.tsp._reserve import reserve_history_gaps + + calls: list[dict] = [] + + async def find_contracts(fqme: str) -> list[str]: + return ['contract'] + + async def schedule(contract: str, **kwargs) -> SimpleNamespace: + calls.append(kwargs) + return mnq_schedule(wider=len(calls) > 1) + + client = SimpleNamespace( + find_contracts=find_contracts, + ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule), + ) + + async def query(left: float, right: float) -> dict: + return await Client.history_schedule( + client, 'mnq.cme.20261218', + datetime.fromtimestamp(left, UTC), + datetime.fromtimestamp(right, UTC), + ) + + frame: np.ndarray = np.zeros(2, dtype=[ + ('time', 'f8'), + ('open', 'f8'), + ('high', 'f8'), + ('low', 'f8'), + ('close', 'f8'), + ]) + frame['time'] = [1790193489, 1790248914] + frame['close'] = [30755.75, 30478.5] + result: np.ndarray + ranges: list[tuple[float, float]] + result, ranges = asyncio.run(reserve_history_gaps( + frame, 1, query, max_rows=51826, + )) + assert calls == [ + { + 'numDays': days, + 'endDateTime': '20260924-11:21:54', + 'useRTH': False, + } + for days in (1, 2) + ] + assert len(result) == 51826 + assert ranges == [ + (1790193490, 1790197200), + (1790200800, 1790248914), + ] + assert not np.any( + (result['time'] >= 1790197200) + & (result['time'] < 1790200800) + ) + np.testing.assert_array_equal(result[[0, -1]], frame) + + +@pytest.mark.parametrize('missing_right', [False, True]) +def test_schedule_retry_preserves_unknown_coverage( + missing_right: bool, +): + ''' + Widening must not turn a partial reply into claimed coverage. + + Return the same incomplete schedule on every fake request. + Missing left coverage exhausts the bounded lookback; missing + right coverage stops immediately because an earlier start cannot + repair it. The unchanged response retains the unknown interval + so downstream reservation still rejects it. + + ''' + calls: list[int] = [] + + async def find_contracts(fqme: str) -> list[str]: + return ['contract'] + + async def schedule(contract: str, **kwargs) -> SimpleNamespace: + calls.append(kwargs['numDays']) + return mnq_schedule() + + client = SimpleNamespace( + find_contracts=find_contracts, + ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule), + ) + end: datetime = datetime(2026, 9, 24, 11, tzinfo=UTC) + if missing_right: + end = datetime(2026, 9, 24, 23, tzinfo=UTC) + out: dict = asyncio.run(Client.history_schedule( + client, 'mnq.cme.20261218', + datetime(2026, 9, 23, 19, tzinfo=UTC), end, + )) + assert out == normalize_schedule(mnq_schedule(), False) + assert calls == ([2] if missing_right else [1, 2, 4, 8, 16, 31]) + + +def test_schedule_retry_shares_timeout( + monkeypatch: pytest.MonkeyPatch, +): + ''' + A retry must not renew the caller's total schedule wait budget. + + Record timeout scopes while the first fake reply lacks coverage + and the second request waits forever. The shared deadline must + cancel that second request and propagate TimeoutError, with no + partial result substituted for failure. No sleep ordering is + required: only the blocked second request yields to the deadline. + + ''' + timeouts: list[float] = [] + calls: int = 0 + cancelled: bool = False + original_timeout = asyncio.timeout + + def timeout(delay: float): + timeouts.append(delay) + return original_timeout(delay) + + monkeypatch.setattr(asyncio, 'timeout', timeout) + + async def find_contracts(fqme: str) -> list[str]: + return ['contract'] + + async def schedule(contract: str, **kwargs) -> SimpleNamespace: + nonlocal calls, cancelled + calls += 1 + if calls == 1: + return mnq_schedule() + try: + await asyncio.Future() + except asyncio.CancelledError: + cancelled = True + raise + + client = SimpleNamespace( + find_contracts=find_contracts, + ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule), + ) + with pytest.raises(TimeoutError): + asyncio.run(Client.history_schedule( + client, 'mnq.cme.20261218', + datetime(2026, 9, 23, 19, tzinfo=UTC), + datetime(2026, 9, 24, 11, tzinfo=UTC), + timeout_s=0.01, + )) + assert timeouts == [0.01] + assert calls == 2 + assert cancelled