Compare commits
3 Commits
749ca1f09c
...
1da9b760a3
| Author | SHA1 | Date |
|---|---|---|
 Gud Boi
|
1da9b760a3 | |
|
Gud Boi
|
efa72980f8 | |
|
Gud Boi
|
5eb4fc9d8a |
|
|
@ -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.
|
||||
|
|
@ -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.
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
---
|
||||
model: openai/gpt-6-astra
|
||||
service: opencode
|
||||
session: unavailable
|
||||
timestamp: 2026-09-16T19:40:52Z
|
||||
git_ref: 749ca1f0
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260916T194052Z_749ca1f0_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Fix wrong IB fill/commission pairing and empty ledger normalization;
|
||||
retain and investigate silent socket loss during network roaming.
|
||||
|
||||
## Response summary
|
||||
|
||||
Preserve execution identity from callbacks into accounting, prevent
|
||||
empty commission IDs from erasing ledger identity, and propagate actual
|
||||
socket disconnection as a transport exception. The existing data-farm
|
||||
reset mechanism is not a socket-session reconnect implementation.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `piker/brokers/ib/broker.py`: fill correlation and socket termination.
|
||||
- `piker/brokers/ib/ledger.py`: authoritative execution ID.
|
||||
- `piker/brokers/ib/api.py`: disconnect event propagation.
|
||||
- `tests/test_ib_trade_events.py`: accounting event regressions.
|
||||
- `tests/test_ib_method_proxy.py`: socket-disconnect regression.
|
||||
|
||||
## Human edits
|
||||
|
||||
The human supplied live traceback and debugger values proving the
|
||||
execution mismatch, requested a regression rather than a KeyError
|
||||
guard, and required continued attention to roaming recovery. No human
|
||||
source edits to this patch have been observed; acceptance is pending.
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
---
|
||||
model: openai/gpt-6-astra
|
||||
service: opencode
|
||||
timestamp: 2026-09-16T19:40:52Z
|
||||
git_ref: 749ca1f0
|
||||
diff_cmd: git diff HEAD -- piker/brokers/ib tests/test_ib_trade_events.py tests/test_ib_method_proxy.py
|
||||
---
|
||||
|
||||
## Human instructions and evidence
|
||||
|
||||
Fix silent IB socket loss after network roaming, reported as
|
||||
`[Errno 104] Connection reset by peer`. Also investigate a live
|
||||
MNQ accounting crash in `emit_pp_update()` with empty normalized
|
||||
ledger entries. The human supplied debugger evidence that a commission
|
||||
for execution 4261 was paired with execution 4262 on the same contract
|
||||
and account, and explicitly requested implementation and regression
|
||||
tests. They asked to retain the roaming task and continue working
|
||||
through follow-up messages.
|
||||
|
||||
## Generated changes
|
||||
|
||||
> `git diff HEAD -- piker/brokers/ib/broker.py`
|
||||
|
||||
Preserve the callback-supplied fill with its commission. Track positions
|
||||
by account and contract ID and pending commissioned fills by execution.
|
||||
Reject mismatched commission/execution IDs. Propagate socket closure
|
||||
instead of returning normally from the trade-event bridge.
|
||||
|
||||
> `git diff HEAD -- piker/brokers/ib/ledger.py`
|
||||
|
||||
Keep Execution.execId authoritative when flattening commission fields.
|
||||
|
||||
> `git diff HEAD -- piker/brokers/ib/api.py`
|
||||
|
||||
Relay the authoritative disconnected event as ConnectionError, retaining
|
||||
the most recent API error text, and raise it from the proxy reader.
|
||||
Automatic session reconnect/resynchronization remains unimplemented.
|
||||
|
||||
> `git diff HEAD -- tests/test_ib_trade_events.py tests/test_ib_method_proxy.py`
|
||||
|
||||
Cover overlapping executions, zero commissions, position-before/after
|
||||
commission ordering, empty commission IDs, and eventkit socket loss.
|
||||
Nine focused tests passed, including the actual eventkit callback
|
||||
bridge and its disconnect exception. Focused Ruff checks and
|
||||
`git diff --check` also passed.
|
||||
|
|
@ -0,0 +1,61 @@
|
|||
---
|
||||
model: openai/gpt-6-astra
|
||||
service: opencode
|
||||
scope: code
|
||||
substantive: true
|
||||
git_ref: 5eb4fc9d
|
||||
raw_file: 20260916_ib_connectivity_integration.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Investigate IB Gateway loss, silent chart survival, infected-actor
|
||||
shutdown errors and leaked shared memory. Integrate pending tagged
|
||||
address and broadcast-consumer compatibility work. The human required
|
||||
isolated worktrees and preservation of the live setup, and selected
|
||||
tractor piker_pin rather than main as the runtime fix base.
|
||||
|
||||
## Response summary
|
||||
|
||||
Created isolated piker and tractor worktrees and initially integrated
|
||||
c07e3cbd and 4af885d9 without committing. That pending merge was later
|
||||
aborted. The current piker scope is the original IB fail-loud patch
|
||||
and its regression tests on 5eb4fc9d; neither upstream history is
|
||||
merged. The live interpreter installation was not changed.
|
||||
|
||||
Integration-era logging and Deribit compatibility fixes were archived
|
||||
outside this patch and are absent from the current worktree. The raw
|
||||
companion preserves that earlier response as historical output.
|
||||
|
||||
## Generated files
|
||||
|
||||
- piker/brokers/ib/api.py: propagate actual socket disconnection with
|
||||
the last API error through the method-proxy relay.
|
||||
- piker/brokers/ib/broker.py: raise on trade-event socket loss and
|
||||
propagate the disconnect event to the trade consumer.
|
||||
- tests/test_ib_method_proxy.py: exercise API error plus disconnect.
|
||||
- tests/test_ib_trade_events.py: preserve callback fill identity and
|
||||
raise when the callback bridge disconnects.
|
||||
|
||||
Existing IB fail-loud edits and their tests were copied from the
|
||||
earlier session, whose provenance remains in the fill commit.
|
||||
The commit plan includes these four files and both provenance logs.
|
||||
|
||||
## Verification
|
||||
|
||||
Historically, twenty selected piker regression cases passed on the
|
||||
integrated tree against isolated, patched tractor sources. Those
|
||||
results do not validate the current post-abort tree or installed
|
||||
pinned tractor. The new exact-tree commit-plan checks remain pending;
|
||||
no project tests were run during this planning pass. No live
|
||||
service/container tests or installed dependency changes were performed.
|
||||
Original live-worktree diff matched SHA256
|
||||
cbdfa439e656ef3a5effaa0f0e1a7706f1070de863fc0e85222d36294eed8524.
|
||||
|
||||
## Human contributions
|
||||
|
||||
The human identified the dependency branches, corrected the proposed
|
||||
tractor main base to piker_pin, and required a new worktree after the
|
||||
agent initially started an uncommitted merge in the live checkout.
|
||||
That merge was aborted and preservation was verified before isolated
|
||||
work continued. The human supplied all live incident evidence.
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
---
|
||||
model: openai/gpt-6-astra
|
||||
service: opencode
|
||||
git_ref: 5eb4fc9d
|
||||
diff_cmd: git diff HEAD
|
||||
---
|
||||
|
||||
> git diff HEAD -- piker/data/feed.py piker/brokers/deribit/api.py piker/brokers/deribit/feed.py
|
||||
|
||||
Use the standard logging predicate for INFO and repair import-time
|
||||
Deribit message struct/annotation compatibility encountered while
|
||||
validating the merged broadcast-consumer baseline.
|
||||
|
||||
The isolated checks pass, including real shm cleanup:
|
||||
|
||||
- All 20 selected piker integration cases pass after the narrow
|
||||
compatibility fixes.
|
||||
- Tractor's EOF tests pass for TCP and UDS, including an idle
|
||||
owner and an unaffected peer.
|
||||
- All four late-cancellation variants unlink their test-owned
|
||||
data/first/last shm segments before parent teardown.
|
||||
|
||||
The original live-worktree diff still matches its saved hash.
|
||||
We're ready for a Gateway-container test of the full actor tree;
|
||||
automatic reconnect is still a separate, unfinished step.
|
||||
|
|
@ -1276,8 +1276,11 @@ class Client:
|
|||
self.ib.errorEvent.connect(push_err)
|
||||
|
||||
api_err = self.ib.client.apiError
|
||||
last_api_error: str = ''
|
||||
|
||||
def report_api_err(msg: str) -> None:
|
||||
nonlocal last_api_error
|
||||
last_api_error = msg
|
||||
with remove_handler_on_err(
|
||||
api_err,
|
||||
report_api_err,
|
||||
|
|
@ -1290,6 +1293,25 @@ class Client:
|
|||
|
||||
api_err.connect(report_api_err)
|
||||
|
||||
disconnected = self.ib.disconnectedEvent
|
||||
|
||||
def report_disconnect() -> None:
|
||||
# IB.disconnectedEvent is authoritative for socket loss;
|
||||
# numeric IB farm/order errors are a different namespace
|
||||
# from OS errno values reported through Client.apiError.
|
||||
disconnected.disconnect(report_disconnect)
|
||||
with remove_handler_on_err(
|
||||
disconnected,
|
||||
report_disconnect,
|
||||
):
|
||||
reason = last_api_error or 'IB API socket disconnected'
|
||||
to_trio.send_nowait((
|
||||
'disconnected',
|
||||
ConnectionError(reason),
|
||||
))
|
||||
|
||||
disconnected.connect(report_disconnect)
|
||||
|
||||
def positions(
|
||||
self,
|
||||
account: str = '',
|
||||
|
|
@ -1851,6 +1873,9 @@ async def relay_client_proxy_messages(
|
|||
# proxy.reset()
|
||||
|
||||
match msg:
|
||||
case ('disconnected', ConnectionError() as err):
|
||||
raise err
|
||||
|
||||
# Correlated response from a method call.
|
||||
case {'mid': _}:
|
||||
proxy._deliver_method_response(msg)
|
||||
|
|
|
|||
|
|
@ -338,7 +338,8 @@ async def recv_trade_updates(
|
|||
|
||||
case 'commissionReportEvent':
|
||||
assert report
|
||||
emit: CommissionReport = report
|
||||
assert fill is not None
|
||||
emit: tuple[Fill, CommissionReport] = (fill, report)
|
||||
|
||||
case 'execDetailsEvent':
|
||||
# execution details event
|
||||
|
|
@ -390,6 +391,7 @@ async def recv_trade_updates(
|
|||
|
||||
# let the engine run and stream
|
||||
await client.ib.disconnectedEvent
|
||||
raise ConnectionError('IB trade-event socket disconnected')
|
||||
|
||||
|
||||
async def update_and_audit_pos_msg(
|
||||
|
|
@ -1189,15 +1191,10 @@ async def deliver_trade_events(
|
|||
Format and relay all trade events for a given client to emsd.
|
||||
|
||||
'''
|
||||
# task local msg dialog tracking
|
||||
clears: dict[
|
||||
Contract,
|
||||
list[
|
||||
IbPosition | None, # filled by positionEvent
|
||||
Fill | None, # filled by order status and exec details
|
||||
]
|
||||
] = {}
|
||||
execid2con: dict[str, Contract] = {}
|
||||
# Positions belong to an account/contract; commissions belong to
|
||||
# executions. Never recover a Fill from just its Contract.
|
||||
positions: dict[tuple[str, int], IbPosition] = {}
|
||||
pending_fills: dict[tuple[str, int], dict[str, Fill]] = {}
|
||||
|
||||
# TODO: for some reason we can receive a ``None`` here when the
|
||||
# ib-gw goes down? Not sure exactly how that's happening looking
|
||||
|
|
@ -1296,91 +1293,28 @@ async def deliver_trade_events(
|
|||
# translating them to ``Status`` msgs) if we can
|
||||
# show the equivalent status events are no more latent.
|
||||
case 'execDetailsEvent':
|
||||
# unpack attrs pep-0526 style.
|
||||
trade: Trade
|
||||
fill: Fill
|
||||
trade, fill = item
|
||||
con: Contract = trade.contract
|
||||
execu: Execution = fill.execution
|
||||
execid: str = execu.execId
|
||||
report: CommissionReport = fill.commissionReport
|
||||
|
||||
# always fill in id to con map so when commissions
|
||||
# arrive we can maybe fire the pos update..
|
||||
execid2con[execid] = con
|
||||
|
||||
# TODO:
|
||||
# - normalize out commissions details?
|
||||
# - this is the same as the unpacking loop above in
|
||||
# ``trades_to_ledger_entries()`` no?
|
||||
|
||||
# 2 cases:
|
||||
# - fill comes first or
|
||||
# - commission report comes first
|
||||
clear: tuple = clears.setdefault(
|
||||
con,
|
||||
[None, fill],
|
||||
)
|
||||
pos, _fill = clear
|
||||
|
||||
# NOTE: we have to handle the case where a pos msg
|
||||
# has already been set (bc we already relayed rxed
|
||||
# one before both the exec-deats AND the
|
||||
# comms-report?) but the comms-report hasn't yet
|
||||
# arrived, so we fill in the fill (XD) and wait for
|
||||
# the cost to show up before relaying the pos msg
|
||||
# to the EMS..
|
||||
if _fill is None:
|
||||
clear[1] = fill
|
||||
|
||||
cost: float = report.commission
|
||||
if (
|
||||
pos
|
||||
and fill
|
||||
and cost
|
||||
):
|
||||
await emit_pp_update(
|
||||
ems_stream,
|
||||
accounts_def,
|
||||
proxies,
|
||||
ledgers,
|
||||
tables,
|
||||
|
||||
ibpos=pos,
|
||||
fill=fill,
|
||||
)
|
||||
clears.pop(con)
|
||||
# Accounting waits for commissionReportEvent, whose
|
||||
# callback supplies the exact Fill, including zero
|
||||
# commissions. Order-status handling relays fills
|
||||
# independently above.
|
||||
continue
|
||||
|
||||
case 'commissionReportEvent':
|
||||
|
||||
cr: CommissionReport = item
|
||||
fill, cr = item
|
||||
execid: str = cr.execId
|
||||
|
||||
# only fire a pp msg update if,
|
||||
# - we haven't already
|
||||
# - the fill event has already arrived
|
||||
# but it didn't yet have a commision report
|
||||
# which we fill in now.
|
||||
|
||||
# placehold i guess until someone who know wtf
|
||||
# contract this is from can fill it in...
|
||||
con: Contract | None = execid2con.setdefault(execid, None)
|
||||
if (
|
||||
con
|
||||
and (clear := clears.get(con))
|
||||
):
|
||||
pos, fill = clear
|
||||
if (
|
||||
pos
|
||||
and fill
|
||||
):
|
||||
now_cr: CommissionReport = fill.commissionReport
|
||||
if (now_cr != cr):
|
||||
log.warning(
|
||||
'UhhHh ib updated the commission report mid-fill..?\n'
|
||||
f'was: {pformat(cr)}\n'
|
||||
f'now: {pformat(now_cr)}\n'
|
||||
if execid != fill.execution.execId:
|
||||
raise ValueError(
|
||||
f'IB commission/fill execution mismatch:\n'
|
||||
f'{cr!r}\n{fill!r}'
|
||||
)
|
||||
key = (
|
||||
fill.execution.acctNumber,
|
||||
fill.contract.conId,
|
||||
)
|
||||
fill = fill._replace(commissionReport=cr)
|
||||
if (pos := positions.get(key)) is None:
|
||||
pending_fills.setdefault(key, {})[execid] = fill
|
||||
continue
|
||||
|
||||
await emit_pp_update(
|
||||
ems_stream,
|
||||
|
|
@ -1388,13 +1322,9 @@ async def deliver_trade_events(
|
|||
proxies,
|
||||
ledgers,
|
||||
tables,
|
||||
|
||||
ibpos=pos,
|
||||
fill=fill,
|
||||
)
|
||||
clears.pop(con)
|
||||
# TODO: should we clean this?
|
||||
# execid2con.pop(execid)
|
||||
|
||||
# always update with latest ib pos msg info since
|
||||
# we generally audit against it for sanity and
|
||||
|
|
@ -1407,9 +1337,17 @@ async def deliver_trade_events(
|
|||
bs_mktid, ppmsg = pack_position(pos, accounts_def)
|
||||
log.info(f'New IB position msg: {ppmsg}')
|
||||
|
||||
_, fill = clears.setdefault(
|
||||
con,
|
||||
[pos, None],
|
||||
key = (pos.account, con.conId)
|
||||
positions[key] = pos
|
||||
for fill in pending_fills.pop(key, {}).values():
|
||||
await emit_pp_update(
|
||||
ems_stream,
|
||||
accounts_def,
|
||||
proxies,
|
||||
ledgers,
|
||||
tables,
|
||||
ibpos=pos,
|
||||
fill=fill,
|
||||
)
|
||||
# only send a pos update once we've actually rxed
|
||||
# the msg from IB since generally speaking we use
|
||||
|
|
@ -1424,6 +1362,9 @@ async def deliver_trade_events(
|
|||
ibpos=pos,
|
||||
)
|
||||
|
||||
case 'disconnected':
|
||||
raise item
|
||||
|
||||
case 'error':
|
||||
# NOTE: see impl deats in
|
||||
# `Client.inline_errors()::push_err()`
|
||||
|
|
|
|||
|
|
@ -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'
|
||||
|
|
|
|||
|
|
@ -429,9 +429,16 @@ def api_trades_to_ledger_entries(
|
|||
for attr_name, val in fdict.items():
|
||||
match attr_name:
|
||||
# value is a `@dataclass` subtype
|
||||
case 'contract' | 'execution' | 'commissionReport':
|
||||
case 'contract' | 'execution':
|
||||
txn_dict.update(asdict(val))
|
||||
|
||||
case 'commissionReport':
|
||||
report_fields = asdict(val)
|
||||
# Execution.execId owns ledger identity. An empty
|
||||
# CommissionReport.execId must not erase it.
|
||||
report_fields.pop('execId')
|
||||
txn_dict.update(report_fields)
|
||||
|
||||
case 'time':
|
||||
# ib has wack ns timestamps, or is that us?
|
||||
continue
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ IB asyncio method-proxy regressions.
|
|||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
from eventkit import Event
|
||||
import pytest
|
||||
import trio
|
||||
|
||||
|
|
@ -63,6 +64,33 @@ class FakeChannel:
|
|||
await self._tx.aclose()
|
||||
|
||||
|
||||
def test_socket_disconnect_raises_from_proxy() -> None:
|
||||
'''
|
||||
Roaming reset the IB socket, but apiError was merely logged and
|
||||
proxy callers could wait indefinitely. Emit the actual eventkit
|
||||
apiError/disconnected sequence through Client.inline_errors().
|
||||
The single-reader proxy must raise ConnectionError carrying the
|
||||
reset reason, while the preceding error alone remains nonfatal.
|
||||
|
||||
'''
|
||||
async def main() -> None:
|
||||
chan = FakeChannel()
|
||||
ib = SimpleNamespace(
|
||||
errorEvent=Event('errorEvent'),
|
||||
disconnectedEvent=Event('disconnectedEvent'),
|
||||
client=SimpleNamespace(apiError=Event('apiError')),
|
||||
)
|
||||
client = ib_api.Client(ib=ib, config={})
|
||||
client.inline_errors(chan._tx)
|
||||
proxy = MethodProxy(chan, {}, asyncio_ns=client)
|
||||
ib.client.apiError.emit('[Errno 104] Connection reset by peer')
|
||||
ib.disconnectedEvent.emit()
|
||||
with pytest.raises(ConnectionError, match='Errno 104'):
|
||||
await relay_client_proxy_messages(chan, proxy)
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_method_proxy_routes_after_idle_status_events() -> None:
|
||||
'''
|
||||
Idle IB status traffic must not break first-symbol qualification.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,160 @@
|
|||
'''
|
||||
Execution identity across IB fill and commission callbacks.
|
||||
|
||||
'''
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
from bidict import bidict
|
||||
from eventkit import Event
|
||||
from ib_async import (
|
||||
CommissionReport,
|
||||
Contract,
|
||||
Execution,
|
||||
Fill,
|
||||
Position,
|
||||
)
|
||||
import pytest
|
||||
import trio
|
||||
|
||||
from piker.brokers.ib import broker
|
||||
from piker.brokers.ib.ledger import api_trades_to_ledger_entries
|
||||
|
||||
|
||||
def make_fill(tid: str) -> Fill:
|
||||
return Fill(
|
||||
Contract(conId=815824267, secType='FUT', symbol='MNQ'),
|
||||
Execution(
|
||||
execId=tid,
|
||||
acctNumber='DU_TEST',
|
||||
time=1789586554.0,
|
||||
),
|
||||
CommissionReport(),
|
||||
1789586554.0,
|
||||
)
|
||||
|
||||
|
||||
def test_execution_id_survives_empty_commission() -> None:
|
||||
'''
|
||||
Flattening a default CommissionReport erased Execution.execId,
|
||||
causing a real fill to be skipped as an ID-less adjustment. Pass
|
||||
a fill with a valid execution and empty commission through ledger
|
||||
conversion and assert that its account and execution survive.
|
||||
|
||||
'''
|
||||
entries = api_trades_to_ledger_entries(
|
||||
bidict({'DU_TEST': 'ib.algopaper'}),
|
||||
[make_fill('execution-1')],
|
||||
)
|
||||
assert entries['ib.algopaper']['execution-1']['execId'] == (
|
||||
'execution-1'
|
||||
)
|
||||
|
||||
|
||||
def test_callback_preserves_fill_and_disconnect() -> None:
|
||||
'''
|
||||
The asyncio callback dropped commissionReportEvent's exact Fill,
|
||||
forcing the Trio side to guess by contract. Exercise the actual
|
||||
eventkit relay and assert it preserves the supplied object. After
|
||||
that callback, emit disconnectedEvent: the bridge must raise
|
||||
rather than return normally and leave its Trio owner asleep.
|
||||
One asyncio checkpoint lets recv_trade_updates install handlers
|
||||
before the test producer emits either event.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
ib = SimpleNamespace(**{
|
||||
name: Event(name)
|
||||
for name in (
|
||||
'orderStatusEvent', 'commissionReportEvent',
|
||||
'execDetailsEvent', 'positionEvent',
|
||||
'disconnectedEvent',
|
||||
)
|
||||
})
|
||||
emitted = []
|
||||
chan = SimpleNamespace(
|
||||
started_nowait=lambda _: None,
|
||||
send_nowait=emitted.append,
|
||||
)
|
||||
client = SimpleNamespace(ib=ib, inline_errors=lambda _: None)
|
||||
fill = make_fill('A')
|
||||
report = CommissionReport(execId='A', commission=.61)
|
||||
|
||||
async def produce():
|
||||
await asyncio.sleep(0)
|
||||
ib.commissionReportEvent.emit(None, fill, report)
|
||||
ib.disconnectedEvent.emit()
|
||||
|
||||
producer = asyncio.create_task(produce())
|
||||
with pytest.raises(ConnectionError, match='socket'):
|
||||
await broker.recv_trade_updates(chan, client)
|
||||
await producer
|
||||
assert emitted == [('commissionReportEvent', (fill, report))]
|
||||
|
||||
asyncio.run(main())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('position_first', [True, False])
|
||||
def test_overlapping_execution_commissions(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
position_first: bool,
|
||||
) -> None:
|
||||
'''
|
||||
Two executions on one contract shared a single pending Fill.
|
||||
A commission for execution A could therefore account execution B.
|
||||
Queue both executions before either commission, delivering fees
|
||||
in reverse order, and exercise positions before and after fees.
|
||||
The deterministic channel order verifies every execution reaches
|
||||
accounting once with its own fee, including a zero commission.
|
||||
|
||||
'''
|
||||
fills = [make_fill('A'), make_fill('B')]
|
||||
pos = Position('DU_TEST', fills[0].contract, 2, 58000)
|
||||
recorded = []
|
||||
|
||||
async def capture(*args, ibpos, fill=None):
|
||||
if fill is not None:
|
||||
recorded.append((
|
||||
fill.execution.execId,
|
||||
fill.commissionReport.execId,
|
||||
fill.commissionReport.commission,
|
||||
ibpos.account,
|
||||
))
|
||||
|
||||
monkeypatch.setattr(broker, 'emit_pp_update', capture)
|
||||
monkeypatch.setattr(
|
||||
broker, 'pack_position', lambda *args: ('815824267', {})
|
||||
)
|
||||
|
||||
async def main():
|
||||
send, recv = trio.open_memory_channel(10)
|
||||
events = [
|
||||
('execDetailsEvent', (
|
||||
SimpleNamespace(contract=fill.contract), fill,
|
||||
))
|
||||
for fill in fills
|
||||
]
|
||||
events.extend([
|
||||
('commissionReportEvent', (
|
||||
fills[1], CommissionReport(execId='B', commission=0),
|
||||
)),
|
||||
('commissionReportEvent', (
|
||||
fills[0], CommissionReport(execId='A', commission=.61),
|
||||
)),
|
||||
])
|
||||
events.insert(0 if position_first else len(events), (
|
||||
'positionEvent', pos,
|
||||
))
|
||||
async with send, recv:
|
||||
for event in events:
|
||||
send.send_nowait(event)
|
||||
await send.aclose()
|
||||
await broker.deliver_trade_events(
|
||||
recv, None, {}, {}, {}, {}, None,
|
||||
)
|
||||
|
||||
trio.run(main)
|
||||
assert recorded == [
|
||||
('B', 'B', 0, 'DU_TEST'),
|
||||
('A', 'A', .61, 'DU_TEST'),
|
||||
]
|
||||
Loading…
Reference in New Issue