.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: 1da9b760a3
Source-author: goodboy <goodboy_foss@protonmail.com>
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`))
wkt/land_ib_recovery
parent
6f626a9c45
commit
02bb193f87
|
|
@ -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'
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue