pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 11 Packages Projects Releases Wiki Activity

Repair IB startup gaps in `ShmArray` 

Keep reverse history and stored SHM layout aligned so sparse IB
frames do not expose permanent zero-time reservations.

Deats,
- omit inclusive query endpoints from SHM but retain full storage
- serialize and bound farm resets, blanks, and cancellations
- run the final null sweep after reverse retrieval completes
- require valid timestamp boundaries for synthetic gap rows
- cover the live QQQ 653-row failure and reset interleavings

Prompt-IO: ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
backfiller_deep_fixes
Gud Boi 2026-07-29 16:46:52 -04:00
parent ad6c560f6b
commit 0846cbd4a4
7 changed files with 1025 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.claude/skills/run-tests/test-harness-reference.md
View File

@ -107,6 +107,7 @@ Deterministic or local first-pass targets:
- `tests/test_storage_audit.py`
- `tests/test_backfill_audit_snippet.py`
- `tests/test_ib_history.py`
- `tests/test_history_backfill.py`
- `tests/test_ldshm.py`
- `tests/test_accounting.py::test_account_file_default_empty`
- `tests/test_services.py::test_runtime_boot`
@ -191,6 +192,7 @@ tests/
test_ems.py actor, EMS, and paper-position behavior
test_feeds.py live Binance/Kraken feeds and shared memory
test_ib_history.py deterministic IB history request formatting
test_history_backfill.py deterministic history/SHM orchestration
test_ldshm.py ldshm unpublished-slot guard
test_questrade.py obsolete credentialed tests; skipped
test_services.py pikerd/datad/feed/EMS actor lifecycle
@ -207,7 +209,8 @@ tests/
| `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped |
| `piker/storage/_audit.py`, `piker/storage/cli.py` | `tests/test_storage_audit.py` | direct Typer app, no actor |
| `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only |
| `piker/brokers/ib/api.py` history formatting | `tests/test_ib_history.py` | fake client, no network |
| `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network |
| `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM |
| `piker/storage/cli.py` ldshm null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
| `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |
| `piker/accounting/` | targeted accounting node | some cases use live/configured state |

37
ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.md 100644
View File

@ -0,0 +1,37 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: 253ade33-07ba-45b5-8839-be35d17ec164
timestamp: 2026-07-29T20:40:46Z
git_ref: ad6c560f
scope: code
substantive: true
raw_file: 20260729T204046Z_ad6c560f_prompt_io.raw.md
---
## Prompt
The user reported that the QQQ 1-second chart still contained a gap after
the `ldshm` fix, asked the agent to inspect it, and authorized continued
implementation after a short pause and shutdown of the live workspace.
## Response summary
Diagnosed a 653-row active RT SHM null reservation caused by inclusive IB
reverse-query overlaps and startup repair orchestration. Prevented extra
physical SHM overlap rows, serialized and bounded IB history recovery,
made blank/error responses terminate, and constrained synthetic repair to
valid interior timestamp boundaries. Added deterministic concurrency and
data-layout regressions.
## Files changed
- `piker/brokers/ib/feed.py` - bound and serialize IB history reset/retry behavior.
- `piker/tsp/_history.py` - prevent SHM overlap skew and safely finalize null repair.
- `tests/test_ib_history.py` - cover blank, cancellation, reset, and pacing interleavings.
- `tests/test_history_backfill.py` - cover overlap and null-repair invariants.
- `.claude/skills/run-tests/test-harness-reference.md` - register deterministic coverage.
## Human edits
None - generated output remains uncommitted.

72
ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.raw.md 100644
View File

@ -0,0 +1,72 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-07-29T20:40:46Z
git_ref: ad6c560f
diff_cmd: git diff HEAD~1..HEAD
---
The user reported that QQQ still displayed a gap on the 1-second chart
after the focused `ldshm` fix and asked the agent to inspect it. The user
then paused and resumed implementation while shutting down the live piker
trading workspace.
Read-only live inspection found the active datad generation held 653
contiguous zero-time rows in QQQ's 1-second RT SHM, while durable Parquet
contained no zero rows. The run occupied absolute indexes 168148 through
168800 between timestamps 1785348398 and 1785349051. The stale actor
generation and active 60-second history were separately identified.
> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
Generated history changes that exclude each inclusive reverse-query
endpoint already present in SHM while retaining the full provider frame
for NativeDB merge. Startup null repair now runs after reverse backfill,
uses bounded UI notifications, handles provider no-data without an
interactive pause, fills only exact interior zero groups, and refuses
synthetic timestamps unless both valid boundaries establish the expected
cadence.
> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py`
Generated IB history reset changes that make blank responses complete,
bound repeated cancellation and reset failures per request, serialize
timeout and pacing farm resets with one actor-global Trio lock, carry
explicit reset results, and coordinate pacing handoff with queued timeout
waiters.
> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
Generated regressions for inclusive endpoint exclusion, full NativeDB
provider deltas, strict synthetic gap boundaries, empty-provider local
fallback, bounded notifications, and preservation of valid margin rows.
> `git diff HEAD~1..HEAD -- tests/test_ib_history.py`
Generated deterministic regressions for blank IB responses, cancellation
limits, stalled reset cancellation, pacing reset failure limits,
cross-request reset serialization, timeout-to-pacing handoff, and
successful handoff timeout restart.
> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
Registered the deterministic history-backfill target and expanded the IB
history mapping to include `feed.py`.
Verification output:
```text
........................................................................ [ 93%]
..... [100%]
77 passed in 2.12s
```
`git diff --check` passed. Multiple adversarial review rounds identified
and drove fixes for unbounded IB empty/cancellation/reset paths, stale
reset ownership, actor-global failure poisoning, concurrent farm resets,
double-counted pacing handoffs, physical endpoint overlap, and unsafe
synthetic timestamp boundaries. Final review reported no findings.
Live validation was intentionally deferred because the user shut down the
trading workspace. A fresh datad/chart startup is required to regenerate
SHM and verify that QQQ no longer contains the 653-row null run.

111
piker/brokers/ib/feed.py
View File

@ -32,7 +32,6 @@ import time
from typing import (
Any,
Callable,
TYPE_CHECKING,
)
from async_generator import aclosing
@ -75,9 +74,6 @@ from ._util import (
)
from .symbols import get_mkt_info
if TYPE_CHECKING:
from trio._core._run import Task
log = get_logger(
name=__name__,
)
@ -353,6 +349,7 @@ async def wait_on_data_reset(
tuple[
trio.CancelScope,
trio.Event,
list[bool],
]
] = trio.TASK_STATUS_IGNORED,
) -> bool:
@ -396,8 +393,9 @@ async def wait_on_data_reset(
client: Client = proxy._aio_ns
done = trio.Event()
result: list[bool] = []
with trio.move_on_after(timeout) as cs:
task_status.started((cs, done))
task_status.started((cs, done, result))
log.warning(
'Sending DATA RESET request:\n'
@ -413,6 +411,7 @@ async def wait_on_data_reset(
'NO VNC DETECTED!\n'
'Manually press ctrl-alt-f on your IB java app'
)
result.append(False)
done.set()
return False
@ -427,6 +426,7 @@ async def wait_on_data_reset(
]:
await ev.wait()
log.info(f"{name} DATA RESET")
result.append(True)
done.set()
return True
@ -435,12 +435,12 @@ async def wait_on_data_reset(
'Data reset task canceled?'
)
result.append(False)
done.set()
return False
_data_resetter_task: Task | None = None
_failed_resets: int = 0
_data_reset_lock = trio.Lock()
async def get_bars(
@ -479,8 +479,11 @@ async def get_bars(
`.ib.api.Client` methods.
'''
global _data_resetter_task, _failed_resets
nodatas_count: int = 0
cancelled_count: int = 0
failed_resets: int = 0
pacing_reset_pending: bool = False
pacing_reset_completed: bool = False
data_cs: trio.CancelScope | None = None
result: tuple[
@ -493,10 +496,16 @@ async def get_bars(
async def query():
global _failed_resets
nonlocal result, data_cs, end_dt, nodatas_count
nonlocal result, data_cs, end_dt
nonlocal nodatas_count, cancelled_count, failed_resets
nonlocal pacing_reset_pending, pacing_reset_completed
while _failed_resets < max_failed_resets:
while True:
if failed_resets >= max_failed_resets:
raise DataUnavailable(
f'IB history reset failed {failed_resets} '
f'times for {fqme}'
)
try:
(
bars,
@ -528,6 +537,10 @@ async def get_bars(
)
# NOTE: REQUIRED to pass back value..
result = None
failed_resets = 0
result_ready.set()
if data_cs:
data_cs.cancel()
return None
# not enough bars signal, likely due to venue
@ -587,6 +600,7 @@ async def get_bars(
first_dt,
last_dt,
)
failed_resets = 0
# signal data reset loop parent task
result_ready.set()
@ -642,6 +656,12 @@ async def get_bars(
'Query cancelled by IB (:eyeroll:):\n'
f'{err.message}'
)
cancelled_count += 1
if cancelled_count >= max_failed_resets:
raise DataUnavailable(
f'IB cancelled {cancelled_count} history '
f'queries for {fqme}'
)
continue
elif (
@ -662,6 +682,7 @@ async def get_bars(
)
client = proxy._aio_ns.ib.client
pacing_reset_pending = True
# cancel any existing reset task
if data_cs:
@ -669,69 +690,81 @@ async def get_bars(
data_cs.cancel()
# spawn new data reset task
data_cs, reset_done = await tn.start(
async with _data_reset_lock:
(
data_cs,
reset_done,
reset_result,
) = await tn.start(
partial(
wait_on_data_reset,
proxy,
reset_type='connection'
)
)
if reset_done:
_failed_resets = 0
await reset_done.wait()
if reset_result[0]:
failed_resets = 0
else:
_failed_resets += 1
failed_resets += 1
pacing_reset_pending = False
pacing_reset_completed = True
continue
else:
raise
# TODO: make this global across all history task/requests
# such that simultaneous symbol queries don't try data resettingn
# too fast..
unset_resetter: bool = False
async with (
tractor.trionics.collapse_eg(),
trio.open_nursery() as tn
):
# start history request that we allow
# to run indefinitely until a result is acquired
# Run history until a result or bounded reset failure.
tn.start_soon(query)
# start history reset loop which waits up to the timeout
# for a result before triggering a data feed reset.
# Serialize farm resets across every history requester.
while not result_ready.is_set():
with trio.move_on_after(feed_reset_timeout):
await result_ready.wait()
break
if _data_resetter_task:
# don't double invoke the reset hack if another
# requester task already has it covered.
async with _data_reset_lock:
if result_ready.is_set():
break
if pacing_reset_completed:
pacing_reset_completed = False
continue
else:
_data_resetter_task = trio.lowlevel.current_task()
unset_resetter: bool = True
# spawn new data reset task
data_cs, reset_done = await tn.start(
(
data_cs,
reset_done,
reset_result,
) = await tn.start(
partial(
wait_on_data_reset,
proxy,
reset_type='data',
)
)
# sync wait on reset to complete
await reset_done.wait()
_data_resetter_task = (
None
if unset_resetter
else _data_resetter_task
if (
result_ready.is_set()
or
reset_result[0]
):
failed_resets = 0
elif pacing_reset_pending:
# Pacing recovery deliberately cancelled this reset
# and will consume the retry itself.
pass
else:
failed_resets += 1
if failed_resets >= max_failed_resets:
raise DataUnavailable(
f'IB history reset failed '
f'{failed_resets} times for {fqme}'
)
assert result
return (
result,
data_cs is not None,

177
piker/tsp/_history.py
View File

@ -206,6 +206,34 @@ async def notify_backfill(
)
def _synthetic_gap_times(
start_t: float,
right_t: float|None,
count: int,
timeframe: float,
) -> np.ndarray|None:
'''
Build bounded synthetic timestamps without crossing valid data.
'''
times: np.ndarray = (
start_t
+
np.arange(1, count + 1)*timeframe
)
if right_t is None:
return None
if not np.isclose(
right_t,
start_t + (count + 1)*timeframe,
):
return None
return times
async def shm_push_in_between(
shm: ShmArray,
to_push: np.ndarray,
@ -262,11 +290,11 @@ async def maybe_fill_null_segments(
task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
) -> list[Frame]:
) -> None:
task_status.started()
frame: Frame = shm.array
frame: Frame = shm.array.copy()
# TODO, put in parent task/daemon root!
import greenback
@ -297,6 +325,7 @@ async def maybe_fill_null_segments(
await tractor.pause()
break
try:
(
array,
next_start_dt,
@ -306,6 +335,16 @@ async def maybe_fill_null_segments(
start_dt=start_dt,
end_dt=end_dt,
)
except (
NoData,
DataUnavailable,
) as exc:
log.warning(
f'Provider could not repair SHM gap:\n'
f'{start_dt} -> {end_dt}\n'
f'{exc}\n'
)
break
if array.size == 0:
log.warning(
@ -314,7 +353,6 @@ async def maybe_fill_null_segments(
)
# ?TODO? do we want to remove the nulls and push
# the close price here for the gap duration?
await tractor.pause()
break
if (
@ -327,7 +365,7 @@ async def maybe_fill_null_segments(
f'frame_start_dt: {frame_start_dt!r}\n'
f'backfill_until_dt: {backfill_until_dt!r}\n'
)
await tractor.pause()
break
# XXX TODO: pretty sure if i plot tsla, btcusdt.binance
# and mnq.cme.ib this causes a Qt crash XXDDD
@ -349,21 +387,11 @@ async def maybe_fill_null_segments(
# - remember that in the display side, only refersh this
# if the respective history is actually "in view".
# loop
try:
await sampler_stream.send({
'broadcast_all': {
# XXX NOTE XXX: see the
# `.ui._display.increment_history_view()` if block
# that looks for this info to FORCE a hard viz
# redraw!
'backfilling': (mkt.fqme, timeframe),
},
})
except tractor.ContextCancelled:
# log.exception
await tractor.pause()
raise
await notify_backfill(
sampler_stream,
mkt,
timeframe,
)
# RECHECK for more null-gaps
frame: Frame = shm.array
@ -379,7 +407,7 @@ async def maybe_fill_null_segments(
(
iabs_slices,
iabs_zero_rows,
zero_t,
_zero_t,
) = null_segs
log.warning(
f'{len(iabs_slices)} NULL TIME SEGMENTS DETECTED!\n'
@ -398,13 +426,61 @@ async def maybe_fill_null_segments(
'close',
]
for istart, istop in iabs_slices:
zero_indexes: np.ndarray = np.asarray(iabs_zero_rows)
split_at: np.ndarray = (
np.flatnonzero(np.diff(zero_indexes) != 1)
+
1
)
zero_groups: list[np.ndarray] = np.split(
zero_indexes,
split_at,
)
repaired: int = 0
frame_start: int = int(frame['index'][0])
for zero_group in zero_groups:
istart: int = int(zero_group[0])
istop: int = int(zero_group[-1]) + 1
seed_index: int = istart - 1
if seed_index < frame_start:
log.warning(
f'Can not forward-fill leading SHM nulls:\n'
f'{istart}:{istop}\n'
)
continue
# get view into buffer for null-segment
seed: np.void = shm._array[seed_index]
start_t: float = seed['time']
if start_t <= 0:
log.warning(
f'Can not forward-fill SHM nulls from '
f'zero-time seed at {seed_index}\n'
)
continue
# Fill only the null rows, preserving both valid margins.
gap: np.ndarray = shm._array[istart:istop]
cls: float = seed['close']
# copy the oldest OHLC samples forward
cls: float = shm._array[istart]['close']
frame_end: int = int(frame['index'][-1])
right_t: float|None = (
shm._array[istop]['time']
if istop <= frame_end
else None
)
gap_times: np.ndarray|None = _synthetic_gap_times(
start_t,
right_t,
len(gap),
timeframe,
)
if gap_times is None:
log.warning(
f'Can not safely forward-fill SHM nulls:\n'
f'left={start_t} right={right_t} '
f'rows={len(gap)} period={timeframe}\n'
)
continue
# TODO: how can we mark this range as being a gap tho?
# -[ ] maybe pg finally supports nulls in ndarray to
@ -413,30 +489,13 @@ async def maybe_fill_null_segments(
# another col/field to denote?
gap[ohlc_fields] = cls
start_t: float = shm._array[istart]['time']
t_diff: float = (istop - istart)*timeframe
gap['time'] = np.arange(
start=start_t,
stop=start_t + t_diff,
step=timeframe,
)
gap['time'] = gap_times
repaired += len(gap)
# TODO: reimpl using the new `.ui._remote_ctl` ctx
# ideally using some kinda decent
# tractory-reverse-lookup-connnection from some other
# `Context` type thingy?
await sampler_stream.send({
'broadcast_all': {
# XXX NOTE XXX: see the
# `.ui._display.increment_history_view()` if block
# that looks for this info to FORCE a hard viz
# redraw!
'backfilling': (mkt.fqme, timeframe),
},
})
# TODO: interatively step through any remaining
# time-gaps/null-segments and spawn piecewise backfiller
# tasks in a nursery?
@ -447,6 +506,13 @@ async def maybe_fill_null_segments(
# -[ ] fill algo: do queries in alternating "latest, then
# earliest, then latest.. etc?"
if repaired:
await notify_backfill(
sampler_stream,
mkt,
timeframe,
)
async def start_backfill(
get_hist,
@ -680,7 +746,12 @@ async def start_backfill(
array,
prepend_until_dt=backfill_until_dt,
)
to_push: np.ndarray = to_store
# The requested end bar is already the oldest published
# sample. Keep it for idempotent storage merge, but do not
# consume another physical SHM row for the overlap.
to_push: np.ndarray = to_store[
to_store['time'] < last_start_dt.timestamp()
]
ln: int = len(to_push)
if ln:
log.info(
@ -1314,27 +1385,25 @@ async def tsdb_backfill(
log.info(f'Loaded {to_push.shape} datums from storage')
# NOTE: ASYNC-conduct tsdb timestamp gap detection and backfill any
# seemingly missing (null-time) segments..
# TODO: ideally these can never exist!
# 2nd nursery END
if last_tsdb_dt is not None:
# Repair only after reverse backfill releases providers
# such as IB which permit one history request at a time.
# TODO: ideally these null segments can never exist!
# -[ ] somehow it seems sometimes we're writing zero-ed
# segments to tsdbs during teardown?
# -[ ] can we ensure that the backfiller tasks do this
# work PREVENTAVELY instead?
# -[ ] fill in non-zero epoch time values ALWAYS!
# await maybe_fill_null_segments(
await tn.start(partial(
maybe_fill_null_segments,
await maybe_fill_null_segments(
shm=shm,
timeframe=timeframe,
get_hist=get_hist,
sampler_stream=sampler_stream,
mkt=mkt,
backfill_until_dt=last_tsdb_dt,
))
# 2nd nursery END
)
# TODO: maybe start history anal and load missing "history
# gaps" via backend..

185
tests/test_history_backfill.py
View File

@ -29,6 +29,8 @@ from piker.storage.nativedb import (
ohlc_key_map,
)
from piker.tsp._history import (
_synthetic_gap_times,
maybe_fill_null_segments,
notify_backfill,
publish_latest_frame,
start_backfill,
@ -92,15 +94,22 @@ def test_empty_frame_completes() -> None:
def test_storage_receives_full_provider_delta() -> None:
'''
SHM capacity truncation does not truncate durable history.
SHM drops the published endpoint without truncating storage.
Reverse history queries include their requested end bar, which is
already the oldest sample in SHM. Publishing the overlap consumed
an extra physical row and made reserved null counts disagree with
elapsed timestamps. Return a frame with the overlap and prove
NativeDB receives the complete provider delta while SHM receives
only bars older than its published boundary.
'''
frame = np.zeros(
2,
3,
dtype=np.dtype(def_iohlcv_fields),
)
frame['index'] = [0, 1]
frame['time'] = [60, 120]
frame['index'] = [0, 1, 2]
frame['time'] = [60, 120, 180]
events: list[str] = []
@ -142,7 +151,7 @@ def test_storage_receives_full_provider_delta() -> None:
return (
frame,
from_timestamp(60),
from_timestamp(120),
from_timestamp(180),
)
async def main() -> None:
@ -154,7 +163,7 @@ def test_storage_receives_full_provider_delta() -> None:
mkt=mkt,
shm=shm,
timeframe=60,
backfill_from_shm_index=1,
backfill_from_shm_index=2,
backfill_from_dt=from_timestamp(180),
sampler_stream=Sampler(),
backfill_until_dt=from_timestamp(60),
@ -163,8 +172,8 @@ def test_storage_receives_full_provider_delta() -> None:
)
trio.run(main)
assert shm.pushed[0]['time'].tolist() == [120]
assert storage.frames[0]['time'].tolist() == [60, 120]
assert shm.pushed[0]['time'].tolist() == [60, 120]
assert storage.frames[0]['time'].tolist() == [60, 120, 180]
assert events == ['storage', 'shm', 'sampler']
@ -344,3 +353,163 @@ def test_backfill_notification_timeout_is_bounded(
)
trio.run(main)
def test_synthetic_gap_times_require_valid_right_boundary() -> None:
'''
Synthetic rows must remain strictly between valid boundary bars.
The live QQQ reservation held one more physical zero row than its
timestamp interval could represent. Blind fill would assign the
null the right boundary's timestamp, creating a duplicate. Prove
aligned bounds produce cadence while an overfull segment
is rejected instead of inventing non-monotonic timestamps.
'''
aligned = _synthetic_gap_times(
start_t=60,
right_t=300,
count=3,
timeframe=60,
)
overfull = _synthetic_gap_times(
start_t=60,
right_t=240,
count=3,
timeframe=60,
)
trailing = _synthetic_gap_times(
start_t=60,
right_t=None,
count=3,
timeframe=60,
)
assert aligned is not None
assert aligned.tolist() == [120, 180, 240]
assert overfull is None
assert trailing is None
def test_null_repair_falls_back_without_debug_pause(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
Empty provider repair must resolve interior zero rows locally.
QQQ startup exposed 653 zero-time rows between stored history and
an IB frame. The null repair queried IB while reverse backfill
occupied its sole history request slot, then could
enter an interactive debug pause before reaching forward-fill.
Arrange an interior null run, return an empty provider frame, and
wedge sampler notification. A short timeout and fail-fast pause
prove local fallback needs no interactive or IPC progress.
The assertions verify only the
zero rows inherit the preceding close and synthesized timestamps;
both valid boundary rows remain unchanged.
'''
monkeypatch.setattr(
'piker.tsp._history._notify_timeout_s',
0.01,
)
backing = np.zeros(
105,
dtype=np.dtype(def_iohlcv_fields),
)
backing['index'] = np.arange(105)
frame = backing[100:105]
frame['index'] = np.arange(100, 105)
frame['time'] = [60, 0, 0, 0, 300]
frame['open'] = [9, 0, 0, 0, 19]
frame['high'] = [11, 0, 0, 0, 21]
frame['low'] = [8, 0, 0, 0, 18]
frame['close'] = [10, 0, 0, 0, 20]
frame['volume'] = [5, 0, 0, 0, 6]
class Shm:
'''
Expose the test frame through the SHM repair interface.
'''
def __init__(self) -> None:
self._array = backing
@property
def array(self) -> np.ndarray:
'''
Return the readable SHM view.
'''
return self._array[100:105]
class Sampler:
'''
Model a backpressured UI notification stream.
'''
async def send(self, msg) -> None:
'''
Block until the bounded notifier cancels this send.
'''
await trio.sleep_forever()
async def get_hist(*args, **kwargs):
'''
Return no provider bars for the requested null segment.
'''
end_dt = kwargs['end_dt']
empty = np.empty(
0,
dtype=np.dtype(def_iohlcv_fields),
)
return empty, end_dt, end_dt
async def fail_pause() -> None:
'''
Reject interactive debugging in normal repair fallbacks.
'''
raise AssertionError('null repair entered tractor.pause()')
monkeypatch.setattr(tractor, 'pause', fail_pause)
async def main() -> None:
import greenback
async def ensure_portal() -> None:
'''
Avoid installing a greenback portal in this unit test.
'''
monkeypatch.setattr(
greenback,
'ensure_portal',
ensure_portal,
)
with trio.fail_after(0.1):
await maybe_fill_null_segments(
shm=Shm(),
timeframe=60,
get_hist=get_hist,
sampler_stream=Sampler(),
mkt=SimpleNamespace(fqme='qqq.nasdaq.ib'),
backfill_until_dt=from_timestamp(60),
)
trio.run(main)
assert frame['time'].tolist() == [60, 120, 180, 240, 300]
for field in ('open', 'high', 'low', 'close'):
assert frame[field].tolist() == [
frame[field][0],
10,
10,
10,
frame[field][-1],
]
assert frame['volume'].tolist() == [5, 0, 0, 0, 6]

509
tests/test_ib_history.py
View File

@ -8,11 +8,19 @@ from datetime import (
)
from types import SimpleNamespace
from pendulum import datetime as pdatetime
from ib_async import RequestError
import numpy as np
from pendulum import (
datetime as pdatetime,
duration,
)
import pytest
import trio
from piker.brokers import DataUnavailable
from piker.brokers.ib import feed as ib_feed
from piker.brokers.ib.api import Client
from piker.brokers.ib.feed import get_bars
@pytest.mark.parametrize(
@ -79,3 +87,502 @@ def test_history_end_datetime_uses_ib_utc_format(
trio.run(main)
assert request['endDateTime'] == expected
def test_empty_history_frame_completes() -> None:
'''
A blank IB response must release the history request slot.
``get_bars()`` previously returned from its query task without
signalling the parent, which then reset the feed forever. That
kept serialized SHM repair from running. Return IB's empty
response and prove the request completes explicitly
with no result instead of relying on a reset timeout.
'''
class Proxy:
'''
Return one empty IB history response.
'''
async def bars(self, **kwargs):
'''
Model ``Client.bars()`` during a venue data gap.
'''
return (
[],
np.empty(0),
duration(seconds=2000),
)
async def main() -> None:
with trio.fail_after(0.1):
result, timedout = await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
)
assert result is None
assert not timedout
trio.run(main)
def test_repeated_ib_cancellation_is_bounded() -> None:
'''
Repeated IB cancellation must not monopolize history forever.
IB can answer a historical request with error 162 cancellation.
The old retry loop had no limit, so reverse backfill could retain
the backend's sole request slot and prevent final null repair.
Always raise that response and prove the configured retry bound
terminates with ``DataUnavailable`` deterministically.
'''
class Proxy:
'''
Cancel every IB history request.
'''
async def bars(self, **kwargs):
'''
Raise IB's historical-query cancellation response.
'''
raise RequestError(
1,
162,
'API historical data query cancelled',
)
async def main() -> None:
with (
trio.fail_after(0.1),
pytest.raises(DataUnavailable),
):
await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
max_failed_resets=2,
)
trio.run(main)
def test_query_error_cancels_stalled_reset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
Exceptional history exit must cancel its in-flight reset task.
A timed-out request acquires the reset lock before launching
IB's VNC reset task. If the query then raises while that task is
blocked, stale work must not retain the reset lock. Force that
ordering with explicit Trio checkpoints, then prove bounded query
failure cancels the reset task before propagating.
'''
reset_cancelled: list[bool] = []
class Proxy:
'''
Delay once, then return IB cancellation.
'''
async def bars(self, **kwargs):
'''
Let the parent acquire the reset lock before failing.
'''
await trio.sleep(0.02)
raise RequestError(
1,
162,
'API historical data query cancelled',
)
async def stalled_reset(
proxy,
reset_type='data',
task_status=trio.TASK_STATUS_IGNORED,
) -> bool:
'''
Publish reset state, then block until sibling failure.
'''
done = trio.Event()
result: list[bool] = []
with trio.CancelScope() as cs:
task_status.started((cs, done, result))
try:
await trio.sleep_forever()
finally:
reset_cancelled.append(True)
return False
monkeypatch.setattr(
ib_feed,
'wait_on_data_reset',
stalled_reset,
)
async def main() -> None:
with (
trio.fail_after(0.1),
pytest.raises(DataUnavailable),
):
await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
feed_reset_timeout=0.001,
max_failed_resets=1,
)
trio.run(main)
assert reset_cancelled == [True]
def test_failed_pacing_resets_reach_retry_limit(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
Failed pacing resets must increment the request's retry count.
The reset API returns an ``Event`` and result box. The old
code tested the always-truthy event and reset the failure counter
after each failure. Return immediate failures and prove two
pacing responses terminate instead of looping forever.
'''
reset_calls: list[bool] = []
class Proxy:
'''
Return a pacing violation for every request.
'''
_aio_ns = SimpleNamespace(
ib=SimpleNamespace(
client=SimpleNamespace(),
),
)
async def bars(self, **kwargs):
'''
Raise IB's pacing response.
'''
raise RequestError(
1,
162,
ib_feed._pacing,
)
async def failed_reset(
proxy,
reset_type='data',
task_status=trio.TASK_STATUS_IGNORED,
) -> bool:
'''
Publish one completed, unsuccessful reset.
'''
done = trio.Event()
result = [False]
reset_calls.append(True)
with trio.CancelScope() as cs:
task_status.started((cs, done, result))
done.set()
return False
monkeypatch.setattr(
ib_feed,
'wait_on_data_reset',
failed_reset,
)
async def main() -> None:
with (
trio.fail_after(0.1),
pytest.raises(DataUnavailable),
):
await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
max_failed_resets=2,
)
trio.run(main)
assert reset_calls == [True, True]
def test_pacing_resets_are_serialized_across_requests(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
Concurrent history requests must never reset IB farms together.
Pacing recovery bypassed reset ownership, allowing 1s and 60s
requests to launch the VNC hack concurrently. Start two pacing
failures in one nursery and hold each fake reset across a Trio
checkpoint. Active-count assertions prove the actor-global lock
serializes reset lifetimes without depending on scheduler timing.
'''
active: int = 0
max_active: int = 0
reset_calls: int = 0
class Proxy:
'''
Return a pacing violation for every request.
'''
_aio_ns = SimpleNamespace(
ib=SimpleNamespace(
client=SimpleNamespace(),
),
)
async def bars(self, **kwargs):
'''
Raise IB's pacing response.
'''
raise RequestError(
1,
162,
ib_feed._pacing,
)
async def failed_reset(
proxy,
reset_type='data',
task_status=trio.TASK_STATUS_IGNORED,
) -> bool:
'''
Hold one failed reset long enough to expose overlap.
'''
nonlocal active, max_active, reset_calls
done = trio.Event()
result: list[bool] = []
with trio.CancelScope() as cs:
task_status.started((cs, done, result))
active += 1
reset_calls += 1
max_active = max(max_active, active)
await trio.sleep(0.01)
active -= 1
result.append(False)
done.set()
return False
monkeypatch.setattr(
ib_feed,
'wait_on_data_reset',
failed_reset,
)
async def request() -> None:
'''
Run one independently bounded history request.
'''
with pytest.raises(DataUnavailable):
await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
max_failed_resets=1,
)
async def main() -> None:
with trio.fail_after(0.1):
async with trio.open_nursery() as nursery:
nursery.start_soon(request)
nursery.start_soon(request)
trio.run(main)
assert reset_calls == 2
assert max_active == 1
def test_pacing_supersedes_timeout_reset_once(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
Pacing handoff must not count cancellation as another failure.
A slow request can trigger timeout recovery immediately before IB
returns a pacing violation. Query-side recovery then deliberately
cancels that data reset and waits for the same actor-global lock.
Reproduce this with blocked first and failed second resets. A
one-attempt limit proves cancellation releases the lock
without aborting before the intended connection reset runs.
'''
reset_types: list[str] = []
class Proxy:
'''
Return pacing only after timeout recovery starts.
'''
_aio_ns = SimpleNamespace(
ib=SimpleNamespace(
client=SimpleNamespace(),
),
)
async def bars(self, **kwargs):
'''
Delay past the reset timeout, then report pacing.
'''
await trio.sleep(0.02)
raise RequestError(
1,
162,
ib_feed._pacing,
)
async def reset(
proxy,
reset_type='data',
task_status=trio.TASK_STATUS_IGNORED,
) -> bool:
'''
Block data reset and fail connection reset immediately.
'''
done = trio.Event()
result: list[bool] = []
reset_types.append(reset_type)
with trio.CancelScope() as cs:
task_status.started((cs, done, result))
if reset_type == 'data':
await trio.sleep_forever()
result.append(False)
done.set()
return False
monkeypatch.setattr(
ib_feed,
'wait_on_data_reset',
reset,
)
async def main() -> None:
with (
trio.fail_after(0.1),
pytest.raises(DataUnavailable),
):
await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
feed_reset_timeout=0.001,
max_failed_resets=1,
)
trio.run(main)
assert reset_types == ['data', 'connection']
def test_completed_pacing_reset_restarts_timeout_wait(
monkeypatch: pytest.MonkeyPatch,
) -> None:
'''
A queued timeout waiter must consume completed pacing recovery.
While query-side pacing recovery holds the reset lock, the outer
timeout task can already queue behind it. Without a completion
handoff, that waiter launches a redundant data reset immediately.
Return a blank frame just after a successful connection reset and
prove no third reset runs before the restarted timeout observes
completion.
'''
reset_types: list[str] = []
class Proxy:
'''
Pace the first query and return blank data on retry.
'''
_aio_ns = SimpleNamespace(
ib=SimpleNamespace(
client=SimpleNamespace(),
),
)
def __init__(self) -> None:
self.calls: int = 0
async def bars(self, **kwargs):
'''
Control the pacing-to-success handoff ordering.
'''
self.calls += 1
if self.calls == 1:
await trio.sleep(0.02)
raise RequestError(
1,
162,
ib_feed._pacing,
)
await trio.sleep(0.0005)
return (
[],
np.empty(0),
duration(seconds=2000),
)
async def reset(
proxy,
reset_type='data',
task_status=trio.TASK_STATUS_IGNORED,
) -> bool:
'''
Cancel timeout recovery and complete pacing recovery.
'''
done = trio.Event()
result: list[bool] = []
reset_types.append(reset_type)
with trio.CancelScope() as cs:
task_status.started((cs, done, result))
if reset_type == 'data':
await trio.sleep_forever()
result.append(reset_type == 'connection')
done.set()
return result[0]
monkeypatch.setattr(
ib_feed,
'wait_on_data_reset',
reset,
)
async def main() -> None:
with trio.fail_after(0.1):
result, _ = await get_bars(
Proxy(),
'qqq.nasdaq.ib',
1,
feed_reset_timeout=0.001,
max_failed_resets=2,
)
assert result is None
trio.run(main)
assert reset_types == ['data', 'connection']