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`))wkt/ib_history_request_bounds
parent
441358e2e7
commit
eed5856622
|
|
@ -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.
|
||||
|
|
@ -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.
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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'})
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue