Compare commits
No commits in common. "8eb85ad99e6b354f730196932f806d1631f5e3d4" and "441358e2e7689e272d726e993df73e6fe1dbc075" have entirely different histories.
8eb85ad99e
...
441358e2e7
|
|
@ -1,45 +0,0 @@
|
|||
---
|
||||
model: gpt-6
|
||||
service: codex
|
||||
timestamp: 2026-09-24T23:24:05.525363+00:00
|
||||
git_ref: 441358e2
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260924T232405Z_441358e2_schedule_coverage_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
User: "do it", approving the proposed bounded wider-window retry
|
||||
for incomplete schedule coverage and its MNQ regression test.
|
||||
Previous live qualification showed a one-day request began after
|
||||
the gap's left edge; a wider query covered it and excluded the
|
||||
one-hour venue closure.
|
||||
|
||||
## Response summary
|
||||
|
||||
Added bounded backwards widening under one timeout, preserving actual
|
||||
coverage and fixed end bounds. Added captured-response allocation,
|
||||
unknown-coverage, and retry-timeout tests, and updated API documentation.
|
||||
39 selected tests and Ruff passed. Live data was not changed.
|
||||
|
||||
## Files changed
|
||||
|
||||
- piker/brokers/ib/api.py — bounded schedule coverage retries.
|
||||
- tests/providers/test_ib_schedule.py — captured MNQ and retry regressions.
|
||||
- docs/history_segment_backfill.rst — retry contract and limits.
|
||||
|
||||
## Human edits
|
||||
|
||||
The user established schedule-driven allocation, explicit unknown
|
||||
coverage, and avoiding reindexing as the design constraints, supplied
|
||||
the MNQ gap, and authorized this focused coverage fix. No direct human
|
||||
source edits occurred during this patch; review is pending.
|
||||
|
||||
## Follow-up review
|
||||
|
||||
The user requested factoring and documenting the repeated 86400
|
||||
conversion. Added piker.calc.SECONDS_PER_DAY, documented elapsed
|
||||
versus calendar/session days once, and reused it in the IB query
|
||||
and reservation span check. Named the client retry limit locally.
|
||||
Schedule tests: 11 passed; Ruff passed after this adjustment.
|
||||
|
|
@ -1,66 +0,0 @@
|
|||
---
|
||||
model: gpt-6
|
||||
service: codex
|
||||
timestamp: 2026-09-24T23:24:05.525363+00:00
|
||||
git_ref: 441358e2
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/providers/test_ib_schedule.py`
|
||||
|
||||
> `git diff HEAD~1..HEAD -- docs/history_segment_backfill.rst`
|
||||
|
||||
Client.history_schedule() widens incomplete left coverage by doubling
|
||||
numDays up to 31 while preserving endDateTime and actual returned
|
||||
coverage. All attempts share an asyncio timeout. Missing right coverage
|
||||
returns unchanged because extending the lookback cannot repair it.
|
||||
Tests replay the captured MNQ schedules through the real client and
|
||||
reservation allocator, check bounded incomplete coverage and retry
|
||||
cancellation under one timeout. Documentation describes these limits.
|
||||
|
||||
## User-facing output
|
||||
|
||||
The captured MNQ case now passes through the real client method
|
||||
and allocator: the request widens from one day to two, then
|
||||
reserves 51,824 slots while leaving the venue closure compressed.
|
||||
|
||||
The tests also cover permanently incomplete replies and a timeout
|
||||
during the retry. I’m running the related regression suites now.
|
||||
|
||||
All 39 selected tests passed, and Ruff passed. The reservation
|
||||
suite needed a retry outside the sandbox because Trio’s local
|
||||
wakeup socket was blocked.
|
||||
|
||||
The fix and regression tests are ready for review. I’m recording
|
||||
the prompt provenance; live SHM and parquet remain unchanged.
|
||||
|
||||
## Validation commands and results
|
||||
|
||||
Worktree-local py313 in the existing Nix environment; import piker
|
||||
resolved inside this worktree.
|
||||
|
||||
- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/providers/test_ib_schedule.py tests/providers/test_history_reservations.py
|
||||
Sandbox: 11 passed then PermissionError creating Trio wakeup socket.
|
||||
Escalated rerun: 22 passed in 1.02s; fake IB, disposable test SHM.
|
||||
- py313/bin/python -m pytest -p no:xonsh -q -x --tb=short --no-header tests/test_ib_history.py
|
||||
17 passed in 0.69s.
|
||||
- ruff check piker/brokers/ib/api.py tests/providers/test_ib_schedule.py
|
||||
All checks passed (existing deprecated configuration warning).
|
||||
- git diff --check: passed.
|
||||
|
||||
No live service restart, live history repair, staging, or commit.
|
||||
|
||||
## Follow-up user review and generated changes
|
||||
|
||||
User: "feels like this same magic # from the prior commit should be
|
||||
factored and commented once? `num_days: int = max(1, ceil(seconds /
|
||||
86400))` other then that all looks decent"
|
||||
|
||||
> `git diff HEAD~1..HEAD -- piker/calc.py piker/tsp/_reserve.py`
|
||||
|
||||
Added SECONDS_PER_DAY for both consumers, with one comment explaining
|
||||
fixed elapsed days versus calendar days and trading sessions. Named
|
||||
the client's 31-day retry limit locally. After this change the
|
||||
schedule suite passed 11 tests and Ruff passed all four Python files.
|
||||
|
|
@ -21,13 +21,7 @@ Schedule queries
|
|||
for UTC epoch bounds. The response includes actual coverage, timezone,
|
||||
regular-hours policy, and half-open trading sessions. This wraps
|
||||
``Client.history_schedule()`` through the existing asyncio method proxy.
|
||||
The first pass limits a query to 31 days. If IB returns coverage
|
||||
starting after the requested left edge, double the lookback up to
|
||||
31 days while retaining the requested end timestamp. All attempts
|
||||
share one 10-second timeout. A missing right edge cannot be repaired
|
||||
by widening backwards; return the actual partial coverage so the
|
||||
allocator leaves that gap unresolved. The same applies when the
|
||||
31-day lookback still leaves the left edge uncovered.
|
||||
The first pass limits a query to 31 days and times out after 10 seconds.
|
||||
Historical queries must match bars' ``useRTH=False`` selection.
|
||||
|
||||
Do not extrapolate sessions outside response coverage. Empty schedules,
|
||||
|
|
|
|||
|
|
@ -363,13 +363,10 @@ class Client:
|
|||
|
||||
Return UTC coverage separately from sessions. Missing coverage
|
||||
is unknown, never evidence that the venue was closed.
|
||||
Match bars()'s extended-hours policy by default. Retry missing
|
||||
left coverage with a wider lookback, up to 31 days, sharing
|
||||
`timeout_s` across schedule requests. Keep the end fixed.
|
||||
Match bars()'s extended-hours policy by default.
|
||||
|
||||
'''
|
||||
from math import ceil
|
||||
from piker.calc import SECONDS_PER_DAY
|
||||
from .venues import normalize_schedule
|
||||
|
||||
if start_dt.tzinfo is None or end_dt.tzinfo is None:
|
||||
|
|
@ -377,43 +374,23 @@ class Client:
|
|||
seconds: float = (
|
||||
end_dt.astimezone(UTC) - start_dt.astimezone(UTC)
|
||||
).total_seconds()
|
||||
# Bound schedule lookback and retries to one elapsed month.
|
||||
max_schedule_days: int = 31
|
||||
if not 0 < seconds <= max_schedule_days * SECONDS_PER_DAY:
|
||||
if not 0 < seconds <= 31 * 86400:
|
||||
raise ValueError('Schedule range must be within 31 days')
|
||||
contract: Contract = (await self.find_contracts(fqme))[0]
|
||||
num_days: int = max(1, ceil(seconds / SECONDS_PER_DAY))
|
||||
end_stamp: str = end_dt.astimezone(UTC).strftime(
|
||||
'%Y%m%d-%H:%M:%S'
|
||||
schedule = await asyncio.wait_for(
|
||||
self.ib.reqHistoricalScheduleAsync(
|
||||
contract,
|
||||
numDays=max(1, ceil(seconds / 86400)),
|
||||
endDateTime=end_dt.astimezone(UTC).strftime(
|
||||
'%Y%m%d-%H:%M:%S'
|
||||
),
|
||||
useRTH=use_rth,
|
||||
),
|
||||
timeout=timeout_s,
|
||||
)
|
||||
# IB can return a single dated session for a one-day query,
|
||||
# even when the requested interval starts in the preceding
|
||||
# session. Widen the lookback, keeping the requested end fixed.
|
||||
# Doubling bounds request count; 31 days caps provider work.
|
||||
# All attempts share one timeout, rather than renewing it.
|
||||
# TODO: explicit cancellation of the underlying IB request
|
||||
# requires a request-id exposing ib_async schedule wrapper.
|
||||
async with asyncio.timeout(timeout_s):
|
||||
while True:
|
||||
schedule = await self.ib.reqHistoricalScheduleAsync(
|
||||
contract,
|
||||
numDays=num_days,
|
||||
endDateTime=end_stamp,
|
||||
useRTH=use_rth,
|
||||
)
|
||||
coverage: dict = normalize_schedule(schedule, use_rth)
|
||||
if (
|
||||
coverage['start'] <= start_dt.timestamp()
|
||||
or
|
||||
coverage['end'] < end_dt.timestamp()
|
||||
or
|
||||
num_days == max_schedule_days
|
||||
):
|
||||
# Widening backwards cannot fix missing coverage
|
||||
# at the right edge. Preserve actual coverage for
|
||||
# the allocator to reject any unresolved interval.
|
||||
return coverage
|
||||
num_days = min(max_schedule_days, num_days * 2)
|
||||
return normalize_schedule(schedule, use_rth)
|
||||
|
||||
async def bars(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -23,11 +23,6 @@ import itertools
|
|||
from bidict import bidict
|
||||
|
||||
|
||||
# Fixed elapsed-day conversion for timestamp arithmetic. Calendar
|
||||
# days may span 23 or 25 hours across DST; trading sessions vary too.
|
||||
SECONDS_PER_DAY: int = 24 * 60 * 60
|
||||
|
||||
|
||||
_mag2suffix = bidict({3: 'k', 6: 'M', 9: 'B'})
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,143 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Validation shared by local clearing queues and actor streams.
|
||||
|
||||
``msgspec.Struct`` construction does not validate field values. New
|
||||
protocol sends validate outbound values first; Tractor independently
|
||||
validates inbound payloads using its scoped decoder.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
from functools import lru_cache
|
||||
from math import isfinite
|
||||
from types import UnionType
|
||||
from typing import (
|
||||
get_args,
|
||||
Self,
|
||||
TypeVar,
|
||||
)
|
||||
|
||||
import msgspec
|
||||
from msgspec import msgpack
|
||||
|
||||
from piker.types import Struct
|
||||
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
|
||||
class ProtocolError(ValueError):
|
||||
'''
|
||||
A clearing peer violated its message direction or phase.
|
||||
|
||||
'''
|
||||
|
||||
|
||||
class Message(
|
||||
Struct,
|
||||
frozen=True,
|
||||
kw_only=True,
|
||||
tag=True,
|
||||
tag_field='type',
|
||||
forbid_unknown_fields=True,
|
||||
):
|
||||
'''
|
||||
Immutable clearing snapshot with an explicit wire discriminator.
|
||||
|
||||
``OrderState.dialogs`` remains mutable in its owning task.
|
||||
Messages contain snapshots rather than references to that
|
||||
registry.
|
||||
|
||||
'''
|
||||
|
||||
def copy(self, update: dict[str, object]|None = None) -> Self:
|
||||
'''
|
||||
Return a validated replacement without mutating the source.
|
||||
|
||||
'''
|
||||
return validate(
|
||||
msgspec.structs.replace(self, **(update or {})),
|
||||
type(self),
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def decoder(spec: type|UnionType) -> msgpack.Decoder:
|
||||
'''
|
||||
Cache the strict msgpack decoder for one payload spec.
|
||||
|
||||
'''
|
||||
return msgpack.Decoder(spec, strict=True)
|
||||
|
||||
|
||||
def validate(value: T, spec: type[T]|UnionType) -> T:
|
||||
'''
|
||||
Check the local object class, then validate and detach its
|
||||
fields.
|
||||
|
||||
A tagged map is wire-equivalent to a Struct. Rejecting
|
||||
dictionaries here enforces the Python API contract before
|
||||
serialization, while the decoder enforces the actual wire schema.
|
||||
|
||||
'''
|
||||
allowed: tuple[type, ...] = get_args(spec) or (spec,)
|
||||
if type(value) not in allowed:
|
||||
name: str = type(value).__name__
|
||||
raise ProtocolError(f'Unexpected {name}; expected {spec}')
|
||||
decoded: T = decoder(spec).decode(msgpack.encode(value))
|
||||
_check_objects(value, decoded)
|
||||
return decoded
|
||||
|
||||
|
||||
def _check_objects(value: object, decoded: object) -> None:
|
||||
'''
|
||||
Reject nested maps masquerading as messages in local Python
|
||||
calls.
|
||||
|
||||
Decimal and tuple wire encodings cannot prove their original
|
||||
Python types. Compare the validated tree with the supplied tree
|
||||
before admitting it to local state. Integer-to-float widening is
|
||||
the sole accepted scalar coercion.
|
||||
|
||||
'''
|
||||
if (
|
||||
type(value) is not type(decoded)
|
||||
and
|
||||
not (
|
||||
type(value) is int and type(decoded) is float
|
||||
)
|
||||
):
|
||||
raise ProtocolError('Local object differs from its schema')
|
||||
if isinstance(decoded, Message):
|
||||
name: str
|
||||
for name in decoded.__struct_fields__:
|
||||
_check_objects(
|
||||
getattr(value, name),
|
||||
getattr(decoded, name),
|
||||
)
|
||||
elif isinstance(decoded, tuple):
|
||||
original: object
|
||||
normalized: object
|
||||
for original, normalized in zip(value, decoded, strict=True):
|
||||
_check_objects(original, normalized)
|
||||
elif isinstance(decoded, float) and not isfinite(decoded):
|
||||
raise ProtocolError('Non-finite clearing number')
|
||||
elif isinstance(decoded, Decimal) and not decoded.is_finite():
|
||||
raise ProtocolError('Non-finite clearing price')
|
||||
|
|
@ -1,210 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Directional clearing streams over Tractor's typed payload decoder.
|
||||
|
||||
The context decorator declares the whole lifecycle. Each stream
|
||||
narrows reception to its peer's messages; startup records and return
|
||||
values cannot appear as streaming events. Phase barriers are checked
|
||||
by the EMS separately from this direction check.
|
||||
|
||||
'''
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager as acm
|
||||
from types import UnionType
|
||||
from typing import Generic, Self, TypeVar
|
||||
|
||||
import tractor
|
||||
import trio
|
||||
|
||||
from piker.types import Struct
|
||||
|
||||
from ._codec import ProtocolError, validate
|
||||
from ._schema import (
|
||||
BrokerdEvent,
|
||||
BrokerdOpenOrder,
|
||||
BrokerdOpenOrdersComplete,
|
||||
BrokerdPosition,
|
||||
BrokerdRequest,
|
||||
BrokerdStartup,
|
||||
validate_startup,
|
||||
)
|
||||
|
||||
|
||||
Rx = TypeVar('Rx')
|
||||
Tx = TypeVar('Tx')
|
||||
|
||||
|
||||
class Stream(Generic[Rx, Tx]):
|
||||
'''
|
||||
Validate outbound objects before the actor transport sees them.
|
||||
|
||||
Tractor checks the wire payload on reception. The additional
|
||||
local check also protects memory-channel test doubles and
|
||||
disallows maps at the Python API boundary.
|
||||
|
||||
'''
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
stream: tractor.MsgStream,
|
||||
incoming: type[Rx]|UnionType,
|
||||
outgoing: type[Tx]|UnionType,
|
||||
) -> None:
|
||||
'''
|
||||
Bind the raw stream and its directional payload specs.
|
||||
|
||||
'''
|
||||
self.raw: tractor.MsgStream = stream
|
||||
self.incoming: type[Rx]|UnionType = incoming
|
||||
self.outgoing: type[Tx]|UnionType = outgoing
|
||||
|
||||
@property
|
||||
def _closed(self) -> bool:
|
||||
'''
|
||||
Expose the underlying stream closure state.
|
||||
|
||||
'''
|
||||
return self.raw._closed
|
||||
|
||||
@property
|
||||
def chan(self) -> tractor.Channel:
|
||||
'''
|
||||
Expose the Tractor channel used by this stream.
|
||||
|
||||
'''
|
||||
return self.raw.chan
|
||||
|
||||
async def send(self, msg: Tx) -> None:
|
||||
'''
|
||||
Validate and detach an outgoing object before IPC.
|
||||
|
||||
'''
|
||||
await self.raw.send(validate(msg, self.outgoing))
|
||||
|
||||
async def receive(self) -> Rx:
|
||||
'''
|
||||
Receive and validate an object against the incoming spec.
|
||||
|
||||
'''
|
||||
return validate(await self.raw.receive(), self.incoming)
|
||||
|
||||
def __aiter__(self) -> Self:
|
||||
'''
|
||||
Iterate the same directional receiving stream.
|
||||
|
||||
'''
|
||||
return self
|
||||
|
||||
async def __anext__(self) -> Rx:
|
||||
'''
|
||||
Translate normal stream completion into iterator completion.
|
||||
|
||||
'''
|
||||
try:
|
||||
return await self.receive()
|
||||
except trio.EndOfChannel:
|
||||
raise StopAsyncIteration from None
|
||||
|
||||
|
||||
@acm
|
||||
async def open_stream(
|
||||
ctx: tractor.Context,
|
||||
incoming: type[Rx]|UnionType,
|
||||
outgoing: type[Tx]|UnionType,
|
||||
) -> AsyncIterator[Stream[Rx, Tx]]:
|
||||
'''
|
||||
Give one receiving task ownership of the directional decoder.
|
||||
|
||||
'''
|
||||
with ctx.pld_rx.limit_plds(spec=incoming):
|
||||
raw: tractor.MsgStream
|
||||
async with ctx.open_stream() as raw:
|
||||
yield Stream(raw, incoming, outgoing)
|
||||
|
||||
|
||||
class BrokerdPhase(Struct):
|
||||
'''
|
||||
Require one startup barrier before accepting outgoing requests.
|
||||
|
||||
``BrokerdPhase.complete`` belongs to the receiving task. Existing
|
||||
orders are legal only before that barrier, including an empty
|
||||
snapshot. Account and broker ownership are checked independently
|
||||
of decoding.
|
||||
|
||||
'''
|
||||
|
||||
startup: BrokerdStartup
|
||||
complete: bool = False
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
'''
|
||||
Validate the startup snapshot before accepting events.
|
||||
|
||||
'''
|
||||
self.startup = validate_startup(self.startup)
|
||||
|
||||
def receive(self, msg: BrokerdEvent) -> BrokerdEvent:
|
||||
'''
|
||||
Check event ownership and advance the startup barrier once.
|
||||
|
||||
'''
|
||||
msg = validate(msg, BrokerdEvent)
|
||||
if self.startup.mode == 'paper_fallback':
|
||||
raise ProtocolError('Fallback has no live event stream')
|
||||
|
||||
broker: str = self.startup.broker
|
||||
if getattr(msg, 'broker', broker) != broker:
|
||||
raise ProtocolError('Event changed broker ownership')
|
||||
|
||||
if isinstance(msg, BrokerdOpenOrdersComplete):
|
||||
if self.complete:
|
||||
raise ProtocolError('Duplicate startup barrier')
|
||||
self.complete = True
|
||||
return msg
|
||||
if isinstance(msg, BrokerdOpenOrder):
|
||||
if self.complete:
|
||||
raise ProtocolError('Existing order after startup')
|
||||
account: str = msg.order.account
|
||||
symbol: str = msg.order.symbol
|
||||
if not symbol.endswith(f'.{broker}'):
|
||||
raise ProtocolError('Wrong existing-order broker')
|
||||
|
||||
else:
|
||||
account = msg.account
|
||||
if account not in self.startup.accounts:
|
||||
raise ProtocolError('Event has an unannounced account')
|
||||
if isinstance(msg, BrokerdPosition) and not (
|
||||
msg.symbol.endswith(f'.{broker}')
|
||||
):
|
||||
raise ProtocolError('Position has wrong broker')
|
||||
return msg
|
||||
|
||||
def request(self, msg: BrokerdRequest) -> BrokerdRequest:
|
||||
'''
|
||||
Require startup completion and an announced request account.
|
||||
|
||||
'''
|
||||
msg = validate(msg, BrokerdRequest)
|
||||
if not self.complete:
|
||||
raise ProtocolError('Request before startup barrier')
|
||||
|
||||
if msg.account not in self.startup.accounts:
|
||||
raise ProtocolError('Request has an unannounced account')
|
||||
return msg
|
||||
|
|
@ -1,358 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Strict clearing protocol for the staged EMS migration.
|
||||
|
||||
The legacy endpoints still use ``._messages`` until their producers
|
||||
and consumers move together. New endpoints import these schemas;
|
||||
there is deliberately no dictionary compatibility decoder.
|
||||
|
||||
``oid`` belongs to the client/EMS dialog, ``reqid`` to a provider's
|
||||
account, and ``fill_id`` to an execution. Position summaries contain
|
||||
no ledger history. Provider-native records stay inside adapters.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
from typing import Annotated, Literal
|
||||
|
||||
from msgspec import Meta
|
||||
|
||||
from ._codec import Message, ProtocolError, validate
|
||||
|
||||
|
||||
Identity = Annotated[str, Meta(min_length=1)]
|
||||
RequestId = int|Identity
|
||||
TradeSide = Literal['buy', 'sell']
|
||||
OrderResponse = Literal[
|
||||
'pending',
|
||||
'open',
|
||||
'dark_open',
|
||||
'triggered',
|
||||
'closed',
|
||||
'fill',
|
||||
'canceled',
|
||||
'error',
|
||||
]
|
||||
|
||||
|
||||
class Order(Message, tag='order', kw_only=True):
|
||||
'''
|
||||
Client intent for a live order, dark trigger, or alert.
|
||||
|
||||
'''
|
||||
|
||||
oid: Identity
|
||||
symbol: Identity
|
||||
account: Identity
|
||||
action: Literal['buy', 'sell', 'alert']
|
||||
exec_mode: Literal['dark', 'live']
|
||||
price: Decimal
|
||||
size: float # signed: negative sells, positive buys
|
||||
brokers: tuple[Identity, ...] = ()
|
||||
|
||||
|
||||
class Cancel(Message, tag='cancel', kw_only=True):
|
||||
'''
|
||||
Client intent to cancel the dialog identified by ``oid``.
|
||||
|
||||
'''
|
||||
|
||||
oid: Identity
|
||||
symbol: Identity
|
||||
|
||||
|
||||
class BrokerdOrder(Message, tag='broker_order', kw_only=True):
|
||||
'''
|
||||
Submit or edit a provider order for an EMS dialog.
|
||||
|
||||
``BrokerdOrder.reqid`` is absent on a new submission and names
|
||||
the current provider order on an edit. ``BrokerdOrder.symbol``
|
||||
contains the market endpoint without its broker suffix.
|
||||
|
||||
'''
|
||||
|
||||
oid: Identity
|
||||
account: Identity
|
||||
time_ns: int
|
||||
symbol: Identity # broker suffix removed at the EMS boundary
|
||||
price: Decimal
|
||||
size: float
|
||||
action: TradeSide
|
||||
reqid: RequestId|None = None
|
||||
|
||||
|
||||
class BrokerdCancel(Message, tag='broker_cancel', kw_only=True):
|
||||
'''
|
||||
Cancel an acknowledged provider order.
|
||||
|
||||
``BrokerdCancel.reqid`` is required; EMS retains cancellation
|
||||
intent until an acknowledgement establishes that provider ID.
|
||||
|
||||
'''
|
||||
|
||||
oid: Identity
|
||||
account: Identity
|
||||
time_ns: int
|
||||
symbol: Identity
|
||||
reqid: RequestId
|
||||
|
||||
|
||||
class BrokerdOrderAck(Message, tag='broker_ack', kw_only=True):
|
||||
'''
|
||||
Associate an EMS dialog with its account/provider ID.
|
||||
|
||||
'''
|
||||
|
||||
oid: Identity
|
||||
account: Identity
|
||||
reqid: RequestId
|
||||
|
||||
|
||||
class BrokerdStatus(Message, tag='broker_status', kw_only=True):
|
||||
'''
|
||||
Report provider control state independently of executions.
|
||||
|
||||
'''
|
||||
|
||||
time_ns: int
|
||||
reqid: RequestId
|
||||
account: Identity
|
||||
broker: Identity
|
||||
status: Literal['pending', 'open', 'canceled', 'closed']
|
||||
filled: float = 0.0
|
||||
remaining: float = 0.0
|
||||
reason: str = ''
|
||||
|
||||
|
||||
class BrokerdFill(Message, tag='broker_fill', kw_only=True):
|
||||
'''
|
||||
One execution, independently identified from its order.
|
||||
|
||||
``BrokerdFill.broker_time`` is Unix seconds for chart placement.
|
||||
Adapters normalize native clocks and execution IDs before
|
||||
sending. Repeated delivery retains ``BrokerdFill.fill_id``.
|
||||
|
||||
'''
|
||||
|
||||
time_ns: int
|
||||
broker_time: float
|
||||
reqid: RequestId
|
||||
fill_id: Identity
|
||||
account: Identity
|
||||
broker: Identity
|
||||
action: TradeSide
|
||||
size: float
|
||||
price: float
|
||||
|
||||
|
||||
class BrokerdError(Message, tag='broker_error', kw_only=True):
|
||||
'''
|
||||
Report a provider rejection with optional typed request context.
|
||||
|
||||
'''
|
||||
|
||||
account: Identity
|
||||
broker: Identity
|
||||
reason: Identity
|
||||
oid: Identity|None = None
|
||||
reqid: RequestId|None = None
|
||||
symbol: Identity|None = None
|
||||
request: BrokerdOrder|BrokerdCancel|None = None
|
||||
|
||||
|
||||
class BrokerdPosition(Message, tag='position', kw_only=True):
|
||||
'''
|
||||
Account projection; never a durable accounting transaction.
|
||||
|
||||
'''
|
||||
|
||||
broker: Identity
|
||||
account: Identity
|
||||
symbol: Identity # full FQME
|
||||
size: float
|
||||
avg_price: float
|
||||
currency: str = ''
|
||||
bs_mktid: int|str|None = None
|
||||
|
||||
|
||||
class BrokerdOpenOrder(Message, tag='open_order', kw_only=True):
|
||||
'''
|
||||
An existing provider order emitted before the startup barrier.
|
||||
|
||||
``BrokerdOpenOrder.order.oid`` names the provider and account.
|
||||
Its symbol is a full FQME. ``BrokerdOpenOrder.reqid`` retains the
|
||||
provider's control identity.
|
||||
|
||||
'''
|
||||
|
||||
reqid: RequestId
|
||||
order: Order
|
||||
broker: Identity
|
||||
time_ns: int
|
||||
|
||||
|
||||
class BrokerdOpenOrdersComplete(
|
||||
Message,
|
||||
tag='open_orders_complete',
|
||||
kw_only=True,
|
||||
):
|
||||
'''
|
||||
All existing orders have been emitted, including an empty set.
|
||||
|
||||
'''
|
||||
|
||||
broker: Identity
|
||||
|
||||
|
||||
BrokerdRequest = BrokerdOrder|BrokerdCancel
|
||||
BrokerdEvent = (
|
||||
BrokerdOrderAck
|
||||
|BrokerdStatus
|
||||
|BrokerdFill
|
||||
|BrokerdError
|
||||
|BrokerdPosition
|
||||
|BrokerdOpenOrder
|
||||
|BrokerdOpenOrdersComplete
|
||||
)
|
||||
|
||||
|
||||
class Status(Message, tag='status', kw_only=True):
|
||||
'''
|
||||
Immutable EMS snapshot with a typed cause for each transition.
|
||||
|
||||
Mutable cancel and deduplication state belongs to ``Dialog``.
|
||||
Request and event causes occupy separate fields, so neither needs
|
||||
a catch-all dictionary or a recursive copy of provider history.
|
||||
|
||||
'''
|
||||
|
||||
time_ns: int
|
||||
oid: Identity
|
||||
resp: OrderResponse
|
||||
req: Order
|
||||
reqid: RequestId|None = None
|
||||
src: str|None = None
|
||||
cancel_called: bool = False
|
||||
brokerd_request: BrokerdRequest|None = None
|
||||
brokerd_event: BrokerdEvent|None = None
|
||||
|
||||
|
||||
class BrokerdStartup(Message, tag='broker_startup', kw_only=True):
|
||||
'''
|
||||
Announce owned accounts and positions, or delegate to paper.
|
||||
|
||||
'''
|
||||
|
||||
broker: Identity
|
||||
mode: Literal['live', 'paper', 'paper_fallback']
|
||||
accounts: tuple[Identity, ...] = ()
|
||||
positions: tuple[BrokerdPosition, ...] = ()
|
||||
|
||||
|
||||
class EmsStartup(Message, tag='ems_startup', kw_only=True):
|
||||
'''
|
||||
Publish owned positions and active dialogs when EMS is ready.
|
||||
|
||||
'''
|
||||
|
||||
broker: Identity
|
||||
positions: tuple[BrokerdPosition, ...]
|
||||
accounts: tuple[Identity, ...]
|
||||
dialogs: tuple[Status, ...]
|
||||
|
||||
|
||||
EmsRequest = Order|Cancel
|
||||
EmsEvent = Status|BrokerdPosition
|
||||
EmsPayload = EmsStartup|EmsRequest|EmsEvent|None
|
||||
BrokerdPayload = (
|
||||
BrokerdStartup
|
||||
|BrokerdRequest
|
||||
|BrokerdEvent
|
||||
|None
|
||||
)
|
||||
|
||||
|
||||
def validate_startup(msg: BrokerdStartup) -> BrokerdStartup:
|
||||
'''
|
||||
Validate account ownership before publishing a backend snapshot.
|
||||
|
||||
'''
|
||||
msg = validate(msg, BrokerdStartup)
|
||||
_validate_positions(msg.broker, msg.accounts, msg.positions)
|
||||
if msg.mode == 'paper_fallback':
|
||||
if msg.accounts or msg.positions:
|
||||
raise ProtocolError('Fallback cannot publish live state')
|
||||
elif not msg.accounts:
|
||||
raise ProtocolError('Active startup requires an account')
|
||||
elif msg.mode == 'paper' and msg.accounts != ('paper',):
|
||||
raise ProtocolError(
|
||||
'Paper startup requires the paper account'
|
||||
)
|
||||
return msg
|
||||
|
||||
|
||||
def validate_ems_startup(msg: EmsStartup) -> EmsStartup:
|
||||
'''
|
||||
Validate account, position, and active-dialog ownership.
|
||||
|
||||
'''
|
||||
msg = validate(msg, EmsStartup)
|
||||
if not msg.accounts:
|
||||
raise ProtocolError('EMS startup requires an account')
|
||||
broker: str = msg.broker
|
||||
_validate_positions(broker, msg.accounts, msg.positions)
|
||||
oids: set[str] = set()
|
||||
status: Status
|
||||
for status in msg.dialogs:
|
||||
order: Order = status.req
|
||||
if (
|
||||
status.oid != order.oid
|
||||
or status.oid in oids
|
||||
or order.account not in msg.accounts
|
||||
or not order.symbol.endswith(f'.{broker}')
|
||||
or status.resp in ('closed', 'canceled')
|
||||
):
|
||||
raise ProtocolError('Invalid startup dialog ownership')
|
||||
oids.add(status.oid)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_positions(
|
||||
broker: str,
|
||||
accounts: tuple[str, ...],
|
||||
positions: tuple[BrokerdPosition, ...],
|
||||
) -> None:
|
||||
'''
|
||||
Reject duplicate accounts or positions outside their owner.
|
||||
|
||||
'''
|
||||
if len(set(accounts)) != len(accounts):
|
||||
raise ProtocolError('Duplicate startup accounts')
|
||||
keys: set[tuple[str, str]] = set()
|
||||
pos: BrokerdPosition
|
||||
for pos in positions:
|
||||
key: tuple[str, str] = (pos.account, pos.symbol)
|
||||
if (
|
||||
pos.broker != broker
|
||||
or pos.account not in accounts
|
||||
or not pos.symbol.endswith(f'.{broker}')
|
||||
or key in keys
|
||||
):
|
||||
raise ProtocolError('Invalid startup position ownership')
|
||||
keys.add(key)
|
||||
|
|
@ -1,470 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
EMS order-dialog transitions, independent of actor I/O.
|
||||
|
||||
``Account`` owns accounting. ``OrderState`` only correlates order
|
||||
control IDs and produces immutable client snapshots and broker
|
||||
requests. Events can precede an ack or follow a terminal status;
|
||||
finite buffers retain those events without reopening finished orders.
|
||||
|
||||
'''
|
||||
|
||||
from collections import OrderedDict
|
||||
from time import time_ns
|
||||
from typing import Literal
|
||||
|
||||
from msgspec import field
|
||||
from msgspec.structs import replace
|
||||
|
||||
from piker.types import Struct
|
||||
from ._codec import ProtocolError, validate
|
||||
from ._schema import (
|
||||
BrokerdCancel,
|
||||
BrokerdError,
|
||||
BrokerdEvent,
|
||||
BrokerdFill,
|
||||
BrokerdOpenOrder,
|
||||
BrokerdOrder,
|
||||
BrokerdOrderAck,
|
||||
BrokerdRequest,
|
||||
BrokerdStatus,
|
||||
Order,
|
||||
Status,
|
||||
)
|
||||
|
||||
|
||||
class Effects(Struct, frozen=True):
|
||||
'''
|
||||
Return immutable broker requests and client status snapshots.
|
||||
|
||||
'''
|
||||
|
||||
broker: tuple[BrokerdRequest, ...] = ()
|
||||
clients: tuple[Status, ...] = ()
|
||||
|
||||
|
||||
class Dialog(Struct):
|
||||
'''
|
||||
Retain mutable cancellation and fill state for one EMS dialog.
|
||||
|
||||
'''
|
||||
|
||||
status: Status
|
||||
cancel_sent: set[int|str] = field(default_factory=set)
|
||||
fills: set[str] = field(default_factory=set)
|
||||
terminal: bool = False
|
||||
|
||||
|
||||
class OrderState(Struct):
|
||||
'''
|
||||
Own active dialogs, bounded early events and terminal tombstones.
|
||||
|
||||
``OrderState.reqids`` includes the account in each key: separate
|
||||
broker sessions can allocate the same numeric order ID. Old
|
||||
request IDs remain correlated until their tombstone is evicted
|
||||
from ``OrderState.terminal``, so a late execution following an
|
||||
edit still belongs to its original order. Capacity exhaustion
|
||||
raises without discarding early fills.
|
||||
|
||||
'''
|
||||
|
||||
dialogs: dict[str, Dialog] = field(default_factory=dict)
|
||||
active: dict[str, Status] = field(default_factory=dict)
|
||||
reqids: dict[tuple[str, int|str], str] = field(
|
||||
default_factory=dict,
|
||||
)
|
||||
early: dict[tuple[str, int|str], list[BrokerdEvent]] = field(
|
||||
default_factory=dict,
|
||||
)
|
||||
terminal: OrderedDict[str, Dialog] = field(
|
||||
default_factory=OrderedDict,
|
||||
)
|
||||
max_early: int = 256
|
||||
max_terminal: int = 1024
|
||||
|
||||
def register(
|
||||
self,
|
||||
order: Order,
|
||||
resp: Literal['pending', 'open', 'dark_open'] = 'pending',
|
||||
) -> Status:
|
||||
'''
|
||||
Register intent or replace an acknowledged active order.
|
||||
|
||||
Keep ``OrderState.dialogs`` and ``OrderState.active``
|
||||
coherent while rejecting changes to the original order
|
||||
ownership.
|
||||
|
||||
'''
|
||||
order = validate(order, Order)
|
||||
oid: str = order.oid
|
||||
prior: Dialog|None = self.dialogs.get(oid)
|
||||
if oid in self.terminal:
|
||||
raise ProtocolError('Cannot reuse a completed order ID')
|
||||
|
||||
reqid: int|str|None = None
|
||||
if prior:
|
||||
previous: Order = prior.status.req
|
||||
if (
|
||||
previous.symbol != order.symbol
|
||||
or previous.account != order.account
|
||||
or previous.exec_mode != order.exec_mode
|
||||
):
|
||||
raise ProtocolError('Edit changed order ownership')
|
||||
reqid = prior.status.reqid
|
||||
if prior.status.cancel_called:
|
||||
raise ProtocolError('Cannot edit a canceling order')
|
||||
if reqid is None and prior.status.resp != 'dark_open':
|
||||
raise ProtocolError(
|
||||
'Cannot edit before acknowledgement',
|
||||
)
|
||||
status: Status = validate(
|
||||
Status(
|
||||
time_ns=time_ns(),
|
||||
oid=oid,
|
||||
req=order,
|
||||
reqid=reqid,
|
||||
resp=resp,
|
||||
),
|
||||
Status,
|
||||
)
|
||||
if prior:
|
||||
prior.status = status
|
||||
else:
|
||||
self.dialogs[oid] = Dialog(status=status)
|
||||
self.active[oid] = status
|
||||
return status
|
||||
|
||||
def _publish(self, status: Status) -> None:
|
||||
'''
|
||||
Refresh ``Dialog.status`` and ``OrderState.active``.
|
||||
|
||||
'''
|
||||
dialog: Dialog = self.dialogs[status.oid]
|
||||
dialog.status = status
|
||||
if not dialog.terminal:
|
||||
self.active[status.oid] = status
|
||||
|
||||
def _finish(self, status: Status) -> None:
|
||||
'''
|
||||
Move a completed dialog into ``OrderState.terminal``.
|
||||
|
||||
Evict its entries from ``OrderState.reqids`` when terminal
|
||||
retention exceeds ``OrderState.max_terminal``.
|
||||
|
||||
'''
|
||||
oid: str = status.oid
|
||||
dialog: Dialog = self.dialogs.pop(oid)
|
||||
dialog.status = status
|
||||
dialog.terminal = True
|
||||
self.active.pop(oid, None)
|
||||
self.terminal[oid] = dialog
|
||||
while len(self.terminal) > self.max_terminal:
|
||||
old_oid: str
|
||||
_: Dialog
|
||||
old_oid, _ = self.terminal.popitem(last=False)
|
||||
key: tuple[str, int|str]
|
||||
for key in tuple(self.reqids):
|
||||
if self.reqids[key] == old_oid:
|
||||
del self.reqids[key]
|
||||
|
||||
def _cancel(self, dialog: Dialog) -> tuple[BrokerdCancel, ...]:
|
||||
'''
|
||||
Emit a cancel once per ID recorded in ``Dialog.cancel_sent``.
|
||||
|
||||
'''
|
||||
status: Status = dialog.status
|
||||
reqid: int|str|None = status.reqid
|
||||
if (
|
||||
reqid is None
|
||||
or reqid in dialog.cancel_sent
|
||||
or dialog.terminal
|
||||
):
|
||||
return ()
|
||||
dialog.cancel_sent.add(reqid)
|
||||
order: Order = status.req
|
||||
return (
|
||||
BrokerdCancel(
|
||||
oid=status.oid,
|
||||
reqid=reqid,
|
||||
account=order.account,
|
||||
symbol=order.symbol,
|
||||
time_ns=time_ns(),
|
||||
),
|
||||
)
|
||||
|
||||
def cancel(self, oid: str) -> Effects:
|
||||
'''
|
||||
Retain cancel intent or finish an unsubmitted dark order.
|
||||
|
||||
'''
|
||||
dialog: Dialog|None = self.dialogs.get(oid)
|
||||
if dialog is None:
|
||||
return Effects()
|
||||
status: Status = replace(
|
||||
dialog.status,
|
||||
cancel_called=True,
|
||||
time_ns=time_ns(),
|
||||
)
|
||||
if status.resp == 'dark_open':
|
||||
status = replace(status, resp='canceled')
|
||||
self._finish(status)
|
||||
return Effects(clients=(status,))
|
||||
self._publish(status)
|
||||
return Effects(broker=self._cancel(dialog))
|
||||
|
||||
def _correlate(
|
||||
self,
|
||||
account: str,
|
||||
reqid: int|str,
|
||||
oid: str,
|
||||
) -> None:
|
||||
'''
|
||||
Bind account/provider identity in ``OrderState.reqids`` once.
|
||||
|
||||
'''
|
||||
key: tuple[str, int|str] = (account, reqid)
|
||||
previous: str|None = self.reqids.get(key)
|
||||
if previous is not None and previous != oid:
|
||||
raise ProtocolError('Request ID changed ownership')
|
||||
self.reqids[key] = oid
|
||||
|
||||
def accept(self, event: BrokerdEvent) -> Effects:
|
||||
'''
|
||||
Apply one broker event and return immutable transition
|
||||
effects.
|
||||
|
||||
``OrderState.early`` retains uncorrelated events until an ack
|
||||
or startup order establishes their ``OrderState.reqids``
|
||||
entry. ``Dialog.fills`` deduplicates executions without
|
||||
reviving a completed dialog in ``OrderState.active``.
|
||||
|
||||
'''
|
||||
event = validate(event, BrokerdEvent)
|
||||
if isinstance(event, BrokerdOpenOrder):
|
||||
order: Order = event.order
|
||||
if (
|
||||
order.oid in self.dialogs
|
||||
or order.oid in self.terminal
|
||||
):
|
||||
raise ProtocolError('Duplicate startup order')
|
||||
key: tuple[str, int|str] = (order.account, event.reqid)
|
||||
if key in self.reqids:
|
||||
raise ProtocolError('Request ID changed ownership')
|
||||
broker: str = event.broker
|
||||
if not order.symbol.endswith(f'.{broker}'):
|
||||
raise ProtocolError('Wrong existing-order broker')
|
||||
status: Status = self.register(order, resp='open')
|
||||
status = replace(
|
||||
status,
|
||||
reqid=event.reqid,
|
||||
src=event.broker,
|
||||
time_ns=event.time_ns,
|
||||
brokerd_event=event,
|
||||
)
|
||||
self._publish(status)
|
||||
self._correlate(order.account, event.reqid, order.oid)
|
||||
return self._replay(
|
||||
order.account,
|
||||
event.reqid,
|
||||
(status,),
|
||||
)
|
||||
|
||||
if isinstance(event, BrokerdOrderAck):
|
||||
oid: str|None = event.oid
|
||||
dialog: Dialog|None = (
|
||||
self.dialogs.get(oid) or self.terminal.get(oid)
|
||||
)
|
||||
if dialog is None:
|
||||
raise ProtocolError('Ack for an unknown order')
|
||||
if event.account != dialog.status.req.account:
|
||||
raise ProtocolError('Ack changed account ownership')
|
||||
key = (event.account, event.reqid)
|
||||
known: bool = key in self.reqids
|
||||
self._correlate(event.account, event.reqid, oid)
|
||||
if known and dialog.status.reqid != event.reqid:
|
||||
# A delayed duplicate ack for a replaced request ID
|
||||
# cannot revert the current cancel target.
|
||||
return Effects()
|
||||
if not dialog.terminal:
|
||||
self._publish(
|
||||
replace(
|
||||
dialog.status,
|
||||
reqid=event.reqid,
|
||||
time_ns=time_ns(),
|
||||
)
|
||||
)
|
||||
replayed: Effects = self._replay(
|
||||
event.account,
|
||||
event.reqid,
|
||||
)
|
||||
# Replay a terminal event first: an order already closed
|
||||
# before its ack no longer needs a deferred cancel.
|
||||
cancels: tuple[BrokerdCancel, ...] = (
|
||||
self._cancel(dialog)
|
||||
if dialog.status.cancel_called
|
||||
else ()
|
||||
)
|
||||
return Effects(
|
||||
broker=replayed.broker + cancels,
|
||||
clients=replayed.clients,
|
||||
)
|
||||
|
||||
if not isinstance(
|
||||
event,
|
||||
(BrokerdStatus, BrokerdFill, BrokerdError),
|
||||
):
|
||||
raise ProtocolError('Not an order-dialog event')
|
||||
|
||||
account: str = event.account
|
||||
broker: str = event.broker
|
||||
reqid: int|str|None = event.reqid
|
||||
oid = self.reqids.get((account, reqid))
|
||||
if isinstance(event, BrokerdError):
|
||||
error_oid: str|None = event.oid
|
||||
request: BrokerdRequest|None = event.request
|
||||
if request:
|
||||
if (
|
||||
request.account != account
|
||||
or error_oid not in (None, request.oid)
|
||||
):
|
||||
raise ProtocolError(
|
||||
'Error changed request ownership',
|
||||
)
|
||||
error_oid = request.oid
|
||||
if oid and error_oid and oid != error_oid:
|
||||
raise ProtocolError('Error changed order ownership')
|
||||
oid = error_oid or oid
|
||||
if oid is None:
|
||||
if reqid is None:
|
||||
raise ProtocolError('Event has no correlation ID')
|
||||
count: int = sum(map(len, self.early.values()))
|
||||
if count >= self.max_early:
|
||||
raise ProtocolError('Pre-ack capacity exceeded')
|
||||
self.early.setdefault((account, reqid), []).append(event)
|
||||
return Effects()
|
||||
dialog = self.dialogs.get(oid) or self.terminal.get(oid)
|
||||
if dialog is None:
|
||||
raise ProtocolError('Missing correlated dialog')
|
||||
|
||||
if account != dialog.status.req.account:
|
||||
raise ProtocolError('Event changed account ownership')
|
||||
|
||||
if not dialog.status.req.symbol.endswith(f'.{broker}'):
|
||||
raise ProtocolError('Event changed broker ownership')
|
||||
status = replace(
|
||||
dialog.status,
|
||||
time_ns=time_ns(),
|
||||
brokerd_event=event,
|
||||
src=event.broker,
|
||||
)
|
||||
if isinstance(event, BrokerdFill):
|
||||
if not event.fill_id:
|
||||
raise ProtocolError('Fill has no execution identity')
|
||||
if event.fill_id in dialog.fills:
|
||||
return Effects()
|
||||
dialog.fills.add(event.fill_id)
|
||||
# The event retains its execution's reqid. The snapshot
|
||||
# keeps the current control ID after a broker-side edit.
|
||||
status = replace(status, resp='fill')
|
||||
# A late fill is observable but never reopens its dialog.
|
||||
if not dialog.terminal:
|
||||
self._publish(status)
|
||||
return Effects(clients=(status,))
|
||||
if dialog.terminal:
|
||||
return Effects()
|
||||
if (
|
||||
isinstance(event, BrokerdStatus)
|
||||
and reqid != dialog.status.reqid
|
||||
):
|
||||
# Replacing a provider order may cancel the old ID. Its
|
||||
# delayed status cannot close the replacement dialog.
|
||||
return Effects()
|
||||
if isinstance(event, BrokerdError):
|
||||
status = replace(status, resp='error')
|
||||
if (
|
||||
isinstance(event.request, BrokerdOrder)
|
||||
and event.request.reqid is None
|
||||
and dialog.status.reqid is None
|
||||
):
|
||||
# A rejected new submission cannot receive a useful
|
||||
# ack. Retain a tombstone instead of a pending order.
|
||||
self._finish(status)
|
||||
else:
|
||||
self._publish(status)
|
||||
else:
|
||||
status = replace(
|
||||
status, resp=event.status, reqid=reqid
|
||||
)
|
||||
if event.status in ('closed', 'canceled'):
|
||||
self._finish(status)
|
||||
else:
|
||||
self._publish(status)
|
||||
return Effects(clients=(status,))
|
||||
|
||||
def _replay(
|
||||
self,
|
||||
account: str,
|
||||
reqid: int|str,
|
||||
clients: tuple[Status, ...] = (),
|
||||
) -> Effects:
|
||||
'''
|
||||
Drain correlated events from ``OrderState.early``.
|
||||
|
||||
'''
|
||||
requests: tuple[BrokerdRequest, ...] = ()
|
||||
event: BrokerdEvent
|
||||
for event in self.early.pop((account, reqid), ()):
|
||||
effects: Effects = self.accept(event)
|
||||
requests += effects.broker
|
||||
clients += effects.clients
|
||||
return Effects(broker=requests, clients=clients)
|
||||
|
||||
|
||||
def broker_request(
|
||||
order: Order,
|
||||
reqid: int|str|None = None,
|
||||
) -> BrokerdOrder:
|
||||
'''
|
||||
Remove exactly the final broker suffix for the provider API.
|
||||
|
||||
'''
|
||||
order = validate(order, Order)
|
||||
symbol: str
|
||||
_: str
|
||||
broker: str
|
||||
(
|
||||
symbol,
|
||||
_,
|
||||
broker,
|
||||
) = order.symbol.rpartition('.')
|
||||
if not symbol or not broker:
|
||||
raise ProtocolError('Order requires a full FQME')
|
||||
return validate(
|
||||
BrokerdOrder(
|
||||
oid=order.oid,
|
||||
reqid=reqid,
|
||||
symbol=symbol,
|
||||
time_ns=time_ns(),
|
||||
account=order.account,
|
||||
action=order.action,
|
||||
price=order.price,
|
||||
size=order.size,
|
||||
),
|
||||
BrokerdOrder,
|
||||
)
|
||||
|
|
@ -25,7 +25,6 @@ import math
|
|||
|
||||
import numpy as np
|
||||
|
||||
from piker.calc import SECONDS_PER_DAY
|
||||
from piker.log import get_logger
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
|
@ -103,9 +102,10 @@ async def reserve_history_gaps(
|
|||
inserts: dict[int, np.ndarray] = {}
|
||||
ranges: list[tuple[float, float]] = []
|
||||
budget: int = max_rows
|
||||
# Match IB Client.history_schedule()'s first-pass 31-day bound.
|
||||
# Match IB Client.history_schedule()'s first-pass 31-day bound;
|
||||
# 86400 converts elapsed days to seconds, not local venue days.
|
||||
# TODO: advertise query limits through backend history config.
|
||||
max_schedule_span_s: int = 31 * SECONDS_PER_DAY
|
||||
max_schedule_span_s: int = 31 * 86400
|
||||
for offset in reversed(gaps[-newest_gaps:]):
|
||||
left: float = float(times[offset])
|
||||
right: float = float(times[offset + 1])
|
||||
|
|
|
|||
|
|
@ -1,908 +0,0 @@
|
|||
# Typed EMS and Brokerd Messaging
|
||||
|
||||
## October 2026 implementation baseline
|
||||
|
||||
This revision carries the August 28 design into the September Piker
|
||||
baseline (`eed58566`). Preserve IB execution-ID accounting (`5eb4fc9d`)
|
||||
and socket-disconnect propagation (`efa72980`) while changing payloads.
|
||||
The installed Tractor is `5c98ab1f`; qualification must distinguish
|
||||
that April revision from local `main` (`5cb600a8`) and the September
|
||||
context-fault development stack. Do not test against a dirty upstream
|
||||
checkout without recording its changes.
|
||||
|
||||
The complete contract includes local client queues, EMS startup and
|
||||
service readiness, both order-control streams, quote normalization for
|
||||
dark/paper execution, and typed internal trigger/dialog records.
|
||||
Provider-native APIs and persistent ledger formats retain their own
|
||||
normalization boundaries. Runtime handles are not wire messages.
|
||||
|
||||
Use explicit Struct validation before local state mutation or sending;
|
||||
annotations and Struct construction do not validate field values.
|
||||
Receive decoding proves wire shape, not the sender's Python type: a
|
||||
correctly tagged mapping has the same encoding as a Struct.
|
||||
|
||||
Endpoint-wide unions need separate startup, receive-direction, and
|
||||
final-result checks. Validate RPC arguments explicitly until Tractor
|
||||
issue #514 supplies signature-derived `Start` contracts. Never admit
|
||||
`Any` or raw diagnostic mappings into the clearing wire schema.
|
||||
|
||||
Frozen snapshots must use replacement rather than the inherited
|
||||
`Struct.copy(update=...)`, which mutates its source. Keep mutable EMS
|
||||
dialog state separate from published messages and retain execution IDs
|
||||
for fill deduplication. Position messages remain explicit summaries;
|
||||
they must never become synthetic entries in durable trade accounting.
|
||||
|
||||
Port each boundary's producers and consumers together before enabling
|
||||
its strict receive set. Qualify schema errors, wrong-direction and
|
||||
wrong-phase messages, startup completion, cancel-before-ack, reordered
|
||||
and repeated fills, transport failure, and bounded teardown with
|
||||
offline provider fixtures and real local actors. Live order submission
|
||||
is outside this qualification.
|
||||
|
||||
|
||||
### Qualified dependency baseline and first review boundary
|
||||
|
||||
The first boundary introduces `._schema`, local validation in `._codec`,
|
||||
directional streams and lifecycle checks in `._protocol`, and a pure
|
||||
`._state.OrderState` reducer. Existing endpoints keep their original
|
||||
`._messages` imports until their paired migration. This is infrastructure
|
||||
qualification, not a claim that paper or any live adapter is ported.
|
||||
|
||||
The installed baseline is Python 3.13.14, msgspec 0.19.0, Trio 0.32.0,
|
||||
and Tractor `5c98ab1fb6e2cb7781aa75845c7f323a52853c46`. Its existing
|
||||
payload APIs support this boundary without a dependency update.
|
||||
All eight actor contract tests also pass against clean local Tractor
|
||||
`main` at `5cb600a84b6e9fcac68bad04a487b83640b3e5fb`.
|
||||
|
||||
The first set has 65 schema/state/phase tests and eight real IPC tests.
|
||||
These tests cover protocol fixtures and the new reducer; adapter-level
|
||||
conformance, feed-driven paper clearing, abrupt transport EOF, and
|
||||
other spawn backends remain qualification work for subsequent ports.
|
||||
|
||||
That local main adds `--ll` to Tractor's pytest plugin; Piker's current
|
||||
`tests/conftest.py` adds the same option. The ordinary downstream
|
||||
invocation therefore fails before collection. The isolated API check
|
||||
uses `--noconftest -p tractor._testing.pytest` on the same test file.
|
||||
Resolve that harness collision before upgrading Piker's dependency pin.
|
||||
Do not conflate this isolated API qualification with a whole-suite pass.
|
||||
|
||||
Relevant unmerged Tractor work remains a separate qualification input:
|
||||
`pld_dec_refinements` (`6bd685ad`), `auto_codecs` (`45508252`),
|
||||
`wkt/context_fault_status` (`9547c2f9`), and
|
||||
`wkt/ipc_fault_regressions` (`fe9eb738`). The context-fault worktree has
|
||||
uncommitted transport-error handling changes; it is not a clean test
|
||||
baseline. The IPC-fault branch includes `192411c4` for transport EOF
|
||||
propagation. Typed Start arguments remain upstream work; validate them
|
||||
explicitly when migrating the actual clearing contexts.
|
||||
|
||||
## Purpose
|
||||
|
||||
Replace EMS's dictionary-shaped actor traffic with explicit, tagged
|
||||
`msgspec.Struct` protocols and modern Tractor payload enforcement on both
|
||||
actor boundaries:
|
||||
|
||||
```text
|
||||
order client <-> emsd <-> brokerd.<backend> / paperboi.<backend>
|
||||
```
|
||||
|
||||
The broker boundary is part of the formal protocol, not an implementation
|
||||
detail behind the EMS endpoint. Paper, Binance, Kraken, and IB must all emit
|
||||
the same startup type, consume the same request union, and produce the same
|
||||
event union before strict Tractor payload limits are enabled.
|
||||
|
||||
This plan deliberately normalizes schemas and backend behavior before adding
|
||||
`pld_spec`. Enabling transport validation first would only turn existing
|
||||
status drift, startup-shape drift, and raw dictionary reconstruction into
|
||||
hard actor failures without first defining the intended behavior.
|
||||
|
||||
## Scope
|
||||
|
||||
This work includes:
|
||||
|
||||
- typed client-to-EMS requests and EMS-to-client events;
|
||||
- typed EMS-to-broker requests and broker-to-EMS events;
|
||||
- one typed broker startup payload shared by live and paper endpoints;
|
||||
- one typed EMS startup payload;
|
||||
- stable tagged unions suitable for native `msgspec` decoding;
|
||||
- normalized status, identifier, startup, and metadata semantics;
|
||||
- per-direction `Context.pld_rx.limit_plds()` enforcement;
|
||||
- backend-neutral conformance tests for paper, Binance, Kraken, and IB;
|
||||
- deterministic actor-free tests and real Tractor actor/IPC tests;
|
||||
- bounded handling of acknowledgement, status, cancellation, and teardown
|
||||
races;
|
||||
- removal of raw dictionary dispatch after all producers have migrated.
|
||||
|
||||
This work does not include:
|
||||
|
||||
- typing provider-native websocket or REST payloads inside each backend;
|
||||
- changing the datad/brokerd actor split;
|
||||
- spawning brokerd from any location other than `open_brokerd_dialog()`;
|
||||
- redesigning accounting or replacing `BrokerdPosition` with the full
|
||||
accounting `Position` type in the same patch series;
|
||||
- live order submission in CI;
|
||||
- preserving undocumented dictionary IPC between actors running different
|
||||
Piker revisions. The actor tree is treated as a same-checkout deployment.
|
||||
|
||||
## Current Contract Inventory
|
||||
|
||||
### Client and EMS boundary
|
||||
|
||||
| Location | Startup payload | Stream received | Stream sent | Current issue |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `_emsd_main()` | tuple of positions map, accounts list, dialogs map | dictionaries reconstructed as `Order` or `Cancel` | `Status` and `BrokerdPosition`, often decoded as dictionaries by clients | no payload spec, tuple startup, raw matching, loose nested fields |
|
||||
| `open_ems()` / `OrderClient` | destructures tuple | `Status` and `BrokerdPosition` dictionaries | annotated as `Order | dict`, but implementation assumes struct attributes | public shape and wire shape are conflated |
|
||||
|
||||
### EMS and broker boundary
|
||||
|
||||
| Endpoint | Startup payload | Stream received by backend | Stream sent by backend | Current issue |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| paper `open_trade_dialog()` | `(list[BrokerdPosition], ['paper'])` | raw `BrokerdOrder` / `BrokerdCancel` dictionaries | ack, status, fill, error, position structs | raw dispatch and network-dependent simulator setup |
|
||||
| Binance `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | mixed protocol layers and `fill` status drift |
|
||||
| Kraken `open_trade_dialog()` | `'paper'` sentinel or `(positions, accounts)` | raw request dictionaries | structs plus client-side `Status` for existing orders | `filled` status drift and race workarounds |
|
||||
| IB `open_trade_dialog()` | `(positions, tuple(accounts))` | raw request dictionaries | structs plus client-side `Status` for existing orders | status can race ack and internal event types leak into sequencing |
|
||||
|
||||
The shared broker boundary already exists operationally: all four endpoints
|
||||
are long-lived Tractor contexts opened by `open_brokerd_dialog()`. The upgrade
|
||||
formalizes that de facto interface instead of allowing each backend to define
|
||||
its own wire dialect.
|
||||
|
||||
## Target Protocol
|
||||
|
||||
### Tagging and decoding policy
|
||||
|
||||
Every union member must have an explicit stable tag. Use one common tag field,
|
||||
called `type`, with explicit protocol values such as `order`, `cancel`,
|
||||
`broker_ack`, `status`, `broker_fill`, `broker_error`, `position`, and
|
||||
`open_order`. Broker requests have distinct `broker_order` and
|
||||
`broker_cancel` tags so decoding cannot confuse the two directions. Do not rely on
|
||||
Python class names as wire tags.
|
||||
|
||||
All newly formalized wire structs should use:
|
||||
|
||||
- `tag='<stable value>'` and `tag_field='type'`;
|
||||
- `frozen=True` for wire snapshots;
|
||||
- `kw_only=True` where it improves constructor safety;
|
||||
- immutable tuple defaults; factories for mutable internal state;
|
||||
- finite `Literal` values for protocol enums;
|
||||
- `forbid_unknown_fields=True` from schema introduction; adopt the new
|
||||
schema only when both endpoints can meet it.
|
||||
|
||||
The tag is the dispatch discriminator. Fields such as `action`, `status`, and
|
||||
`resp` retain only domain meaning and are not overloaded as message type tags.
|
||||
Mutable EMS dialog bookkeeping must use a separate internal state type. Do not
|
||||
make a wire struct mutable merely because current code mutates and re-sends the
|
||||
same `Status` or `Order` instance.
|
||||
|
||||
Define shared scalar aliases and use them consistently:
|
||||
|
||||
```python
|
||||
RequestId = int | str
|
||||
FillId = str
|
||||
```
|
||||
|
||||
### Client-to-EMS requests
|
||||
|
||||
Define:
|
||||
|
||||
```python
|
||||
EmsRequest = Order | Cancel
|
||||
```
|
||||
|
||||
Normalize the request structs as follows:
|
||||
|
||||
- `Order.action` is exactly `buy | sell | alert`.
|
||||
- `Order.exec_mode` is exactly `dark | live`.
|
||||
- `Order.oid`, `symbol`, and `account` remain required strings.
|
||||
- `Order.price` remains `Decimal`; tests must prove transport round-tripping.
|
||||
- `Order.size` remains signed `float` for this migration. Resolving the
|
||||
duplicated side/action representation is a later domain change.
|
||||
- `Cancel` has its own tag; `action='cancel'` can be removed once no local UI
|
||||
matching depends on it.
|
||||
- Actor IPC accepts structs only. Any user-facing convenience constructor for
|
||||
mappings must validate and construct a struct before entering the stream.
|
||||
|
||||
### EMS-to-client events
|
||||
|
||||
Define:
|
||||
|
||||
```python
|
||||
EmsEvent = Status | BrokerdPosition
|
||||
```
|
||||
|
||||
Retain `BrokerdPosition` as the wire-level position summary for this series,
|
||||
but document it as shared by the broker and client event protocols. A later
|
||||
accounting patch may rename it to `PositionSummary` without coupling that
|
||||
change to IPC enforcement.
|
||||
|
||||
Normalize `Status` so that:
|
||||
|
||||
- `resp` is exactly `pending | open | dark_open | triggered | fill | closed |
|
||||
canceled | error`;
|
||||
- `req` is a required `Order`, never an unvalidated nested dictionary;
|
||||
an uncorrelated backend error must not fabricate a client dialog;
|
||||
- `brokerd_request` is `BrokerdRequest | None` and records the command that
|
||||
caused the status transition;
|
||||
- `brokerd_event` is `BrokerdEvent | None` and records the backend response;
|
||||
- the current mixed-purpose `brokerd_msg` field is removed after callers have
|
||||
migrated; neither replacement accepts an untyped dictionary;
|
||||
- an error is represented by `Status(resp='error', ...)` or by a distinct
|
||||
tagged client event, not the current incomplete `Error(Status)` subclass;
|
||||
- internal mutable dialog state is not the same object sent repeatedly over
|
||||
the wire. Build a status snapshot when broadcasting so prior history and
|
||||
multi-client delivery cannot be changed by later in-place mutation.
|
||||
|
||||
### EMS-to-broker requests
|
||||
|
||||
Define:
|
||||
|
||||
```python
|
||||
BrokerdRequest = BrokerdOrder | BrokerdCancel
|
||||
```
|
||||
|
||||
Normalize both variants so that:
|
||||
|
||||
- they have explicit `broker_order` and `broker_cancel` protocol tags
|
||||
independent of `action`;
|
||||
- `oid` always identifies the EMS dialog;
|
||||
- `BrokerdOrder.reqid` is `RequestId | None` because allocation policy
|
||||
differs by backend; `BrokerdCancel.reqid` is required, and EMS retains
|
||||
cancel intent until acknowledgement supplies that identity;
|
||||
- account and symbol validation failures produce a typed `BrokerdError`
|
||||
rather than logging and continuing silently;
|
||||
- updates remain `BrokerdOrder` with an existing `reqid`; a separate modify
|
||||
variant is deferred unless conformance tests prove the current distinction
|
||||
ambiguous.
|
||||
|
||||
### Broker-to-EMS events
|
||||
|
||||
Define:
|
||||
|
||||
```python
|
||||
BrokerdEvent = (
|
||||
BrokerdOrderAck
|
||||
| BrokerdStatus
|
||||
| BrokerdFill
|
||||
| BrokerdError
|
||||
| BrokerdPosition
|
||||
| BrokerdOpenOrder
|
||||
| BrokerdOpenOrdersComplete
|
||||
)
|
||||
```
|
||||
|
||||
`BrokerdOpenOrder` replaces the current backend-to-EMS use of client-facing
|
||||
`Status`. It carries the provider request id, account, source backend, and a
|
||||
fully typed embedded `Order`. EMS alone translates this event into a client
|
||||
`Status` and registers it in the dialog book.
|
||||
|
||||
Require `oid` on acknowledgements. Status and fill events identify the
|
||||
order by `(account, reqid)` within their broker session; EMS retains that
|
||||
correlation across edits and terminal events. For an external order, the
|
||||
adapter creates a collision-safe stable `oid` from broker, account, and
|
||||
provider `reqid` before emitting `BrokerdOpenOrder`. Its embedded
|
||||
`Order.oid` carries that value. EMS registers it without allocating a
|
||||
reverse-message identity. Service errors may omit `oid`; a typed rejected
|
||||
request can identify a submission which never received a provider ID.
|
||||
|
||||
`BrokerdFill` has a required `fill_id: FillId`. Use the provider execution id
|
||||
or provider event sequence. The four current implementations must use Binance
|
||||
trade id, Kraken transaction id, IB execution id, and a distinct paper-ledger
|
||||
transaction id respectively. Price/size/time are not a safe deduplication key
|
||||
because two legitimate executions may share them. A future provider with no
|
||||
execution identity must define an adapter session-id plus monotonic sequence
|
||||
and document that deduplication does not survive brokerd restart; it may not
|
||||
silently synthesize identity from equal-valued fill fields.
|
||||
|
||||
Use one broker status vocabulary:
|
||||
|
||||
```python
|
||||
BrokerdOrderState = Literal[
|
||||
'pending',
|
||||
'open',
|
||||
'canceled',
|
||||
'closed',
|
||||
]
|
||||
```
|
||||
|
||||
Fills are events, not status values. A partial execution emits a
|
||||
`BrokerdFill` and may then emit `BrokerdStatus(status='open', ...)`. A complete
|
||||
execution emits the final `BrokerdFill` followed by
|
||||
`BrokerdStatus(status='closed', remaining=0)`. Binance `fill`, Kraken
|
||||
`filled`, and IB provider status names must be translated inside their
|
||||
backend adapters and must never cross the brokerd IPC boundary.
|
||||
|
||||
### Startup payloads
|
||||
|
||||
Replace every tuple and string sentinel with tagged startup structs.
|
||||
|
||||
`BrokerdStartup` contains:
|
||||
|
||||
- `mode: Literal['live', 'paper', 'paper_fallback']`;
|
||||
- `broker: str`;
|
||||
- `accounts: tuple[str, ...]`;
|
||||
- `positions: tuple[BrokerdPosition, ...]`.
|
||||
|
||||
A live endpoint without credentials returns
|
||||
`BrokerdStartup(mode='paper_fallback', broker=..., accounts=(), positions=())`.
|
||||
`open_brokerd_dialog()` then opens the paper endpoint, which returns the same
|
||||
type with `mode='paper'`, its paper account, and persisted positions. The
|
||||
fallback value is a typed delegation result, not an active broker snapshot: it
|
||||
is exempt from account/position and open-order-completion requirements, and
|
||||
its context returns after startup. This removes the raw `'paper'` union branch
|
||||
without hiding fallback behavior.
|
||||
|
||||
Existing open orders remain `BrokerdOpenOrder` stream events because some
|
||||
providers discover them only after authenticated stream subscription. The
|
||||
contract must state that startup positions/accounts are a complete snapshot,
|
||||
while open-order synchronization completes through typed stream events. Every
|
||||
active live or paper backend emits exactly one `BrokerdOpenOrdersComplete`
|
||||
after the initial set, including when it found zero orders, so EMS can
|
||||
distinguish completion from delay and can order initial client readiness
|
||||
deterministically. A `paper_fallback` delegation does not open a stream and
|
||||
does not emit this event; the subsequently opened paper endpoint does.
|
||||
|
||||
`EmsStartup` contains:
|
||||
|
||||
- `broker: str`;
|
||||
- `positions: tuple[BrokerdPosition, ...]`;
|
||||
- `accounts: tuple[str, ...]`;
|
||||
- `dialogs: tuple[Status, ...]`.
|
||||
|
||||
`open_ems()` may continue returning the existing nested lookup maps to callers,
|
||||
but it builds them locally from `EmsStartup`. The public convenience shape is
|
||||
therefore separated from the wire contract.
|
||||
|
||||
Treat `BrokerdStartup` as the brokerd equivalent of datad's `FeedInit`, but do
|
||||
not reproduce `validate_backend()`'s long-lived dictionary compatibility
|
||||
branches. Add a broker initialization validator beside the clearing protocol
|
||||
that checks backend name, unique/non-empty accounts, position account
|
||||
membership, `position.broker == startup.broker`, unique
|
||||
`(broker, account, symbol)` position keys, and mode-specific requirements.
|
||||
Give malformed startup its own registered initialization error so failures
|
||||
retain a useful remote exception type before strict Tractor decoding takes
|
||||
over. Each account name must be non-empty; the accounts collection may be
|
||||
empty only for `paper_fallback`. Paper sessions announce exactly
|
||||
`('paper',)`. EMS readiness also requires at least one account.
|
||||
|
||||
Replace `open_brokerd_dialog()`'s manual probing with one broker order-control
|
||||
endpoint resolver. It may use `get_eps(mod, 'brokerd')` as input, but it must
|
||||
select exactly one of `open_trade_dialog` or the temporary legacy
|
||||
`trades_dialogue` alias and must not confuse unrelated `norm_trade` support
|
||||
with a trading dialog. The runtime validator proves startup; the conformance
|
||||
suite proves request/event behavior. Neither mechanism should import or enable
|
||||
datad modules inside the credentialed broker actor.
|
||||
|
||||
### Payload aliases and Tractor limits
|
||||
|
||||
The endpoint-wide payload aliases include startup, stream traffic, and the
|
||||
final `None` result:
|
||||
|
||||
```python
|
||||
EmsPayload = EmsStartup | EmsRequest | EmsEvent | None
|
||||
BrokerdPayload = BrokerdStartup | BrokerdRequest | BrokerdEvent | None
|
||||
```
|
||||
|
||||
Apply broad endpoint specs and narrow receive-side specs as follows:
|
||||
|
||||
| Actor side | Endpoint `pld_spec` | Narrow receive spec |
|
||||
| --- | --- | --- |
|
||||
| emsd `_emsd_main()` | `EmsPayload` | `EmsRequest` around the client command loop |
|
||||
| order client | inherited from endpoint | `EmsEvent` around trade event receives |
|
||||
| backend/paper `open_trade_dialog()` | `BrokerdPayload` | `BrokerdRequest` around backend request loops |
|
||||
| emsd broker relay | inherited from endpoint | `BrokerdEvent` around broker event receives |
|
||||
|
||||
The endpoint payload spec supplies the broad decoder used for `Started`, stream,
|
||||
and return payloads. `ctx.started(..., validate_pld_spec=True)` can also perform
|
||||
local started-value validation. Ordinary `MsgStream.send()` does not validate
|
||||
locally: malformed stream data raises `tractor.MsgTypeError` when the peer
|
||||
receives/decodes it. `ctx.pld_rx.limit_plds(spec=...)` narrows that receiving
|
||||
direction so a valid message from the wrong direction still fails there.
|
||||
|
||||
Do not add either decorator spec until the corresponding boundary has no
|
||||
remaining legacy sender. The final enforcement must follow the proven pattern
|
||||
in `remote_gap_overlays()` and Tractor's own payload-limiting tests.
|
||||
|
||||
### Provider metadata
|
||||
|
||||
Provider-native diagnostics remain inside adapters and their logs. The
|
||||
clearing wire schema has no raw-data extension point or `Any` field.
|
||||
Source backend, broker timestamp, symbol, account, order ID, request ID,
|
||||
and execution ID have explicit typed fields. Errors can carry a typed
|
||||
request. Position summaries contain no mutable accounting history.
|
||||
|
||||
## Ordering and State Semantics
|
||||
|
||||
Typed messages alone do not repair ordering defects. Define these invariants
|
||||
before enforcement:
|
||||
|
||||
1. A backend sends exactly one `BrokerdOrderAck` for each accepted new order.
|
||||
2. The ack establishes `oid <-> reqid` before status/fill events are exposed
|
||||
to EMS whenever the provider API permits that ordering.
|
||||
3. A backend adapter that receives provider events before it can send the ack
|
||||
buffers them by `reqid` and flushes them after ack.
|
||||
4. EMS keeps a bounded defensive pre-ack buffer because actor scheduling
|
||||
can still expose ordering bugs. The first reducer bounds event count
|
||||
and fails loudly on overflow without dropping retained executions.
|
||||
Add wall-clock expiry in its asynchronous owner when porting EMS;
|
||||
the pure reducer does not claim an age bound.
|
||||
5. A cancel received before ack marks the dialog `cancel_called`. After
|
||||
replaying buffered events, the first ack sends one typed
|
||||
`BrokerdCancel` if the dialog is still active.
|
||||
6. Duplicate ack, fill, and terminal status events are idempotent or rejected
|
||||
deterministically. Fill deduplication uses `fill_id`. They never resurrect
|
||||
a closed dialog.
|
||||
7. `closed` and `canceled` are terminal. A late fill may be recorded and
|
||||
reported according to an explicit test, but it cannot recreate mutable
|
||||
active state.
|
||||
8. Existing external orders enter through `BrokerdOpenOrder` with the backend's
|
||||
stable namespaced `oid` and are registered before broadcast.
|
||||
9. Initial open-order discovery is complete only after
|
||||
`BrokerdOpenOrdersComplete`; client readiness cannot infer completion from
|
||||
an empty or temporarily quiet stream.
|
||||
10. Order status is routed by dialog and FQME. Account authorization and
|
||||
position privacy remain a separately scoped subscription design because
|
||||
`_emsd_main()` currently receives no account authorization input. Tests in
|
||||
this series document the current position broadcast and must not claim
|
||||
unauthorized-account isolation.
|
||||
|
||||
## Implementation Phases
|
||||
|
||||
### Phase 0: Characterize the Existing Boundary
|
||||
|
||||
Add tests that record current intended behavior before restructuring:
|
||||
|
||||
- new order, update, cancel, ack, open, partial fill, close, and position flow;
|
||||
- cancel-before-ack;
|
||||
- status-before-ack;
|
||||
- existing provider order discovery;
|
||||
- bad account, zero size, and backend submission error as typed domain errors;
|
||||
- bad action/tag/field type as protocol decode failures, not `BrokerdError`;
|
||||
- startup with live credentials, no credentials, paper account, persisted
|
||||
positions, and no positions;
|
||||
- Binance startup containing a negative/short position;
|
||||
- IB startup containing at least two accounts with distinct contracts;
|
||||
- zero, one, and multiple existing orders followed by the explicit snapshot
|
||||
completion event;
|
||||
- disconnect while requests or events are in flight.
|
||||
|
||||
Use fake streams and fake provider clients; do not require public market data
|
||||
or credentials. Capture known undesirable outcomes as explicit expected
|
||||
failures only when a same-phase fix would obscure the baseline.
|
||||
|
||||
Gate: every behavior changed later has a deterministic assertion describing
|
||||
the old behavior and the target behavior.
|
||||
|
||||
### Phase 1: Define and Audit the Wire Schemas
|
||||
|
||||
Introduce the strict schema in `piker/clearing/_schema.py` alongside the
|
||||
legacy `._messages` imports. Qualify it independently before moving each
|
||||
producer/consumer pair together. Remove the legacy module after every
|
||||
endpoint and public consumer has migrated; do not add dictionary bridges.
|
||||
|
||||
Replace mutable literal defaults with factories. Resolve the accidental tuple
|
||||
default on `BrokerdStatus.account`. Remove or complete the unusable
|
||||
`Error.from_status()` path. Add explicit startup types and
|
||||
`BrokerdOpenOrder`. Add the broker initialization error/validator without a
|
||||
legacy dictionary fallback.
|
||||
|
||||
Inventory every current mutation of `Order`, `Status`, and embedded requests.
|
||||
Introduce a mutable internal dialog-state record and construct frozen wire
|
||||
snapshots from it. Replace mutation-based `Struct.copy(update=...)` usage with
|
||||
new-instance replacement that is proven to work for frozen structs.
|
||||
|
||||
Add actor-free serialization tests using the same msgpack semantics as
|
||||
Tractor:
|
||||
|
||||
- each concrete type round-trips with exact type equality;
|
||||
- each union decodes to the correct concrete type by tag;
|
||||
- nested `Order`, broker event, position, `Decimal`, and optional
|
||||
identifier fields round-trip;
|
||||
- unknown tag, missing required field, wrong literal, wrong primitive type,
|
||||
malformed nested struct, and duplicate/ambiguous tag fail;
|
||||
- unknown fields fail from the first use of the new schema;
|
||||
- request and event aliases do not accept each other's members.
|
||||
|
||||
Gate: all protocol unions have unique tags and can be decoded without manual
|
||||
dictionary inspection.
|
||||
|
||||
### Phase 2: Build a Backend Conformance Harness
|
||||
|
||||
Create reusable conformance helpers that exercise a backend request handler
|
||||
with a memory-backed stream and fake provider client. The harness supplies the
|
||||
same typed request cases and validates only typed events. Every request and
|
||||
emitted event must be encoded and decoded through its declared union before
|
||||
assertion; `isinstance` checks alone are insufficient because direct
|
||||
`msgspec.Struct` construction does not validate annotated field types.
|
||||
|
||||
The shared suite must verify:
|
||||
|
||||
- startup is one `BrokerdStartup`, including typed fallback-to-paper;
|
||||
- malformed startup fails the shared broker initialization validator;
|
||||
- new order produces one correlated ack;
|
||||
- update preserves `oid` and `reqid`;
|
||||
- cancel uses the correlated request id;
|
||||
- invalid account/size produces `BrokerdError` without killing the dialog;
|
||||
- invalid action/tag/type is rejected by codec or Tractor before the handler;
|
||||
- provider statuses normalize to the four broker states;
|
||||
- partial and complete fills use `BrokerdFill` plus normalized statuses;
|
||||
- emitted positions and existing orders are typed;
|
||||
- every active live/paper startup emits one open-order completion event, even
|
||||
for zero orders; `paper_fallback` delegates without one;
|
||||
- duplicate fills are rejected by `fill_id`, while distinct equal-valued fills
|
||||
remain valid;
|
||||
- provider-native dictionaries never cross the test stream;
|
||||
- cancellation closes all handler tasks and leaves no blocked sender;
|
||||
- one malformed request fails at the protocol boundary once limits are
|
||||
enabled.
|
||||
|
||||
Separate backend-neutral contract assertions from provider translation
|
||||
fixtures. Provider fixtures should be the smallest recorded websocket/API
|
||||
payload required to prove normalization. Include regressions for Binance's
|
||||
tuple-shaped error reason, string fill size, short startup position, and
|
||||
float-vs-`Decimal` open-order price; Kraken's `filled` status and open-order
|
||||
price; and IB multi-account startup association.
|
||||
|
||||
Gate: the harness can fail a deliberately nonconforming fake backend for each
|
||||
contract rule.
|
||||
|
||||
### Phase 3: Port the Paper Engine as the Reference Backend
|
||||
|
||||
Make paper `open_trade_dialog()` the reference implementation of
|
||||
`BrokerdPayload`:
|
||||
|
||||
- return `BrokerdStartup`;
|
||||
- consume `BrokerdRequest` instances directly;
|
||||
- emit only `BrokerdEvent` instances;
|
||||
- remove `BrokerdOrder(**request_msg)` and dictionary pattern dispatch;
|
||||
- allocate a distinct deterministic `fill_id`/transaction id instead of
|
||||
reusing `oid` as `tid`;
|
||||
- separate its order protocol tests from market-feed simulation;
|
||||
- allow deterministic synthetic ticks to drive fill tests without a public
|
||||
feed.
|
||||
|
||||
Do not enable endpoint `pld_spec` yet if live backends share code paths that
|
||||
still send legacy payloads.
|
||||
|
||||
Gate: the complete broker conformance suite passes for paper without network,
|
||||
credentials, sleeps, or teardown suppression.
|
||||
|
||||
### Phase 4: Port Live Backend Adapters Individually
|
||||
|
||||
Port one backend per independently reviewable change. A backend is complete
|
||||
only when its provider translation fixtures and shared conformance suite pass.
|
||||
|
||||
Recommended order:
|
||||
|
||||
1. Binance: replace raw request matching, type startup/fallback, replace
|
||||
existing-order `Status` with `BrokerdOpenOrder`, and map `fill` to fill plus
|
||||
`open/closed` events.
|
||||
2. Kraken: replace raw request matching, type startup/fallback, replace
|
||||
existing-order `Status`, map `filled` to fill plus `closed`, and formalize
|
||||
the too-fast edit/cancel behavior.
|
||||
3. IB: replace raw request indexing, type startup, replace existing-order
|
||||
`Status`, and put an explicit sequencing gate between asyncio provider
|
||||
events and ack publication.
|
||||
|
||||
For every backend, core fields are extracted before constructing the event.
|
||||
The provider's original message stays inside its adapter and logs.
|
||||
Keep brokerd's enabled RPC module set minimal and do not reintroduce
|
||||
datad modules into the credentialed actor.
|
||||
|
||||
Each adapter must preserve a provider execution id as `fill_id`, emit the
|
||||
initial open-order completion event, and populate a stable namespaced `oid` for
|
||||
external orders. Binance must retain negative positions in startup. IB startup
|
||||
must associate each position with the correct account/client rather than loop
|
||||
state from the last account.
|
||||
|
||||
Gate: paper, Binance, Kraken, and IB all pass the same offline conformance
|
||||
contract; no backend emits a dictionary or client-facing `Status` to EMS.
|
||||
|
||||
### Phase 5: Convert EMS Broker Translation and State
|
||||
|
||||
Change `open_brokerd_dialog()` to consume `BrokerdStartup` and return a typed
|
||||
relay object. Remove startup tuple/string branching and
|
||||
`BrokerdPosition(**msg)` reconstruction.
|
||||
|
||||
Remove the current `client_ready` dependency from initial broker-event
|
||||
processing. Start the broker relay as soon as its stream opens, consume and
|
||||
reduce `BrokerdOpenOrder` events through `BrokerdOpenOrdersComplete`, and only
|
||||
then construct `EmsStartup` and call `_emsd_main().started()`. The resulting
|
||||
startup dialogs contain the completed initial snapshot. Events arriving before
|
||||
the first client stream exists remain in dialog state/history for startup or
|
||||
later delivery rather than blocking the broker sender or being discarded.
|
||||
|
||||
Split the current large translation loop into a small typed event reducer plus
|
||||
async effects:
|
||||
|
||||
- dispatch with class patterns, not mapping patterns;
|
||||
- update correlation/dialog state in a deterministic function or narrowly
|
||||
scoped state object;
|
||||
- keep that state mutable internally but emit only new frozen `Status`
|
||||
snapshots;
|
||||
- return typed client events and typed broker requests to send;
|
||||
- keep stream broadcast and logging in the async shell;
|
||||
- implement bounded pre-ack buffering and terminal-state idempotence;
|
||||
- translate `BrokerdOpenOrder` to `Status` in one place;
|
||||
- stop reading routing fields from `broker_details`.
|
||||
|
||||
The reducer must be directly testable with event sequences and no actors.
|
||||
|
||||
Gate: all order-state permutations pass actor-free tests, and
|
||||
`translate_and_relay_brokerd_events()` contains no struct reconstruction from
|
||||
dictionaries.
|
||||
|
||||
### Phase 6: Convert the Client Boundary
|
||||
|
||||
Change `process_client_order_cmds()` to receive `EmsRequest` directly and use
|
||||
class-pattern dispatch. Return `EmsStartup` from `_emsd_main()` and construct
|
||||
the existing `open_ems()` convenience maps client-side.
|
||||
|
||||
Make `OrderClient.send()` and `send_nowait()` accept `EmsRequest` only. If a
|
||||
mapping convenience API is required by a known caller, put it in an explicit
|
||||
constructor that performs local msgspec validation; do not put dictionaries on
|
||||
the actor stream. Audit UI consumers and tests so receive loops work with
|
||||
already-decoded `Status` and `BrokerdPosition` instances.
|
||||
|
||||
Replace `OrderClient.update*()` and UI mutations with construction of a new
|
||||
`Order`. Audit mutations of `Status.req`, `Status.resp`, and embedded orders in
|
||||
order mode and clearing code; all should update internal state or replace a
|
||||
wire value, never mutate a frozen payload.
|
||||
|
||||
Gate: there are no `Order(**cmd)`, `Cancel(**cmd)`, `Status(**msg)`, or
|
||||
`BrokerdPosition(**msg)` conversions on the EMS client boundary.
|
||||
|
||||
### Phase 7: Enable Tractor Payload Enforcement
|
||||
|
||||
Enable one boundary at a time:
|
||||
|
||||
1. Add `BrokerdPayload` to every paper/live `open_trade_dialog()` decorator.
|
||||
2. Limit backend receive scopes to `BrokerdRequest`.
|
||||
3. Limit EMS's broker receive scope to `BrokerdEvent`.
|
||||
4. Add `EmsPayload` to `_emsd_main()`.
|
||||
5. Limit EMS's client receive scope to `EmsRequest`.
|
||||
6. Limit order-client receive scopes to `EmsEvent`.
|
||||
7. Verify every producer satisfies the already-strict wire schemas.
|
||||
|
||||
At each step, test invalid startup, stream, wrong-direction, and return
|
||||
payloads. Assert where decoding occurs, `tractor.MsgTypeError` provenance,
|
||||
context cancellation, and bounded teardown rather than assuming every sender
|
||||
raises locally. Add an actor-free metadata test that each paper/Binance/Kraken/
|
||||
IB endpoint's `_tractor_context_meta['pld_spec']` resolves to `BrokerdPayload`
|
||||
and `_emsd_main()` resolves to `EmsPayload`.
|
||||
|
||||
Gate: no valid wrong-direction type or malformed payload reaches business
|
||||
logic, and malformed traffic cannot wedge the actor tree.
|
||||
|
||||
### Phase 8: Real Actor and Lifecycle Qualification
|
||||
|
||||
Add test-only Tractor context endpoints in a dedicated test fixture module to
|
||||
exercise actual encoding, process boundaries, and cancellation. Do not mock
|
||||
`MsgStream` in this layer.
|
||||
|
||||
The real-actor matrix includes:
|
||||
|
||||
- valid startup and bidirectional exchange for both protocol families;
|
||||
- exact concrete types observed after remote decode;
|
||||
- invalid `Started`, stream send, wrong-direction send, and return payload;
|
||||
- optional local `Started` validation and peer receive-side stream/return
|
||||
validation;
|
||||
- child failure before and after `ctx.started()`;
|
||||
- client cancellation while backend send is blocked;
|
||||
- backend cancellation while EMS waits for an event;
|
||||
- simultaneous stream close and terminal status;
|
||||
- repeated open/close cycles with no stale subscribers, tasks, actors, or SHM;
|
||||
- two clients with distinct dialogs and no cross-delivery of order status;
|
||||
- deterministic ack/status and cancel/ack race barriers;
|
||||
- brokerd restart/reconnect policy, explicitly failing if unsupported rather
|
||||
than hanging.
|
||||
|
||||
Use Trio events/channels as barriers. Do not use timing sleeps to force races.
|
||||
Use bounded `fail_after()` scopes around every scenario. Do not use
|
||||
`run_and_tollerate_cancels()` for these tests; expected cancellation must be
|
||||
asserted precisely.
|
||||
|
||||
Gate: the offline real-actor suite passes repeatedly under the repository's
|
||||
supported Tractor spawn methods and fails loudly within a bounded timeout.
|
||||
|
||||
### Phase 9: Remove Compatibility Paths
|
||||
|
||||
After all strict tests pass:
|
||||
|
||||
- delete mapping-pattern IPC dispatch and `Struct(**wire_dict)` conversions;
|
||||
- delete the `'paper'` startup sentinel and startup tuple handling;
|
||||
- delete backend emission of client `Status`;
|
||||
- delete obsolete status aliases such as `fill` and `filled`;
|
||||
- delete legacy `trades_dialogue` lookup if no backend implements it;
|
||||
- remove stale dictionary type annotations and compatibility comments;
|
||||
- remove expected failures introduced in Phase 0;
|
||||
- update backend author documentation with the protocol aliases and
|
||||
conformance command.
|
||||
|
||||
Gate: searches for the removed patterns are empty or have an explicit local,
|
||||
non-IPC justification.
|
||||
|
||||
## Test Layout
|
||||
|
||||
Use focused flat modules consistent with the existing test tree:
|
||||
|
||||
| Test module | Responsibility | Network |
|
||||
| --- | --- | --- |
|
||||
| `tests/test_ems_messages.py` | union tags, codec round-trips, invalid schemas, strict fields | never |
|
||||
| `tests/test_brokerd_protocol.py` | shared backend conformance and provider translation fixtures | never |
|
||||
| `tests/test_ems_state.py` | reducer transitions, correlation, races, idempotence, routing | never |
|
||||
| `tests/test_ems_ipc.py` | real Tractor actor payload limits and lifecycle | never |
|
||||
| `tests/test_ems.py` | public `open_ems()` behavior, persistence, paper integration | deterministic by default |
|
||||
| backend-specific live qualification | authenticated smoke tests only | opt-in |
|
||||
|
||||
The default CI lane runs every actor-free and offline actor test. Authenticated
|
||||
or public-network checks use explicit markers and are never the only coverage
|
||||
for a protocol behavior.
|
||||
|
||||
## Required Test Matrix
|
||||
|
||||
### Schema validation
|
||||
|
||||
- every valid concrete message;
|
||||
- every startup mode;
|
||||
- `int` and `str` request identifiers;
|
||||
- zero, positive, and negative sizes where semantically permitted;
|
||||
- `Decimal` prices;
|
||||
- optional and populated typed request/event causes;
|
||||
- unknown/missing/wrong tags;
|
||||
- unknown fields rejected from schema introduction;
|
||||
- wrong literal and primitive field types;
|
||||
- malformed nested order/event/position;
|
||||
- a request sent in the event direction and vice versa.
|
||||
|
||||
### Dialog transitions
|
||||
|
||||
- submit -> ack -> pending -> open;
|
||||
- submit -> ack -> fill -> open -> fill -> closed;
|
||||
- submit -> ack -> canceled;
|
||||
- submit -> cancel -> ack -> broker cancel -> canceled;
|
||||
- submit -> provider error;
|
||||
- status -> ack reordered delivery;
|
||||
- fill -> ack reordered delivery;
|
||||
- duplicate ack;
|
||||
- duplicate `fill_id` and distinct equal-valued fills;
|
||||
- duplicate and conflicting terminal statuses;
|
||||
- late fill after close;
|
||||
- update before and after ack;
|
||||
- cancel and update for unknown dialog;
|
||||
- existing external open order;
|
||||
- position update with and without an active order.
|
||||
|
||||
### Backend conformance
|
||||
|
||||
- paper, Binance, Kraken, and IB request acceptance;
|
||||
- account rejection;
|
||||
- provider status normalization;
|
||||
- provider exception to `BrokerdError` conversion;
|
||||
- open-order conversion;
|
||||
- zero-order and populated open-order snapshot completion;
|
||||
- startup account/position conversion;
|
||||
- Binance negative-position startup and IB multi-account association;
|
||||
- provider metadata isolation;
|
||||
- backend handler cancellation and stream closure.
|
||||
|
||||
### Actor semantics
|
||||
|
||||
- valid typed startup and streams;
|
||||
- local `Started` and peer receive-side `MsgTypeError` provenance;
|
||||
- malformed startup cancels context cleanly;
|
||||
- malformed stream traffic cancels both ends cleanly;
|
||||
- caller and callee cancellation;
|
||||
- no swallowed teardown errors;
|
||||
- no actor/task/channel leaks;
|
||||
- repeated execution to expose scheduling-sensitive failures.
|
||||
|
||||
### Routing and isolation
|
||||
|
||||
- event reaches the originating dialog subscribers;
|
||||
- unrelated FQME subscriber does not receive it;
|
||||
- current all-client position broadcast is explicitly characterized as a
|
||||
known limitation pending an account authorization/subscription schema;
|
||||
- disconnected clients are removed from subscriber/dialog tables;
|
||||
- reconnect receives only the documented startup snapshot and active dialogs.
|
||||
|
||||
## CI and Qualification Gates
|
||||
|
||||
Run gates in increasing cost order:
|
||||
|
||||
1. schema and reducer unit tests;
|
||||
2. backend conformance tests with fake provider clients;
|
||||
3. real Tractor IPC tests;
|
||||
4. deterministic paper EMS integration;
|
||||
5. the broader clearing/service regression set;
|
||||
6. repeated race tests under each supported spawn method;
|
||||
7. opt-in live read-only startup for Binance, Kraken, and IB;
|
||||
8. manually authorized live order qualification with minimal size.
|
||||
|
||||
No live credential or public-network result can waive an offline failure.
|
||||
Never place an order as part of default CI.
|
||||
|
||||
## Commit Boundaries
|
||||
|
||||
Keep the implementation reviewable and bisectable:
|
||||
|
||||
1. add the refreshed dependency baseline, strict schemas, pure reducer,
|
||||
lifecycle guards, and offline/actor protocol qualification;
|
||||
2. characterize and port paper together with its EMS consumer;
|
||||
3. port each live adapter and its paired EMS handling with provider
|
||||
translation fixtures and execution-identity tests;
|
||||
4. migrate client/EMS and UI consumers together;
|
||||
5. enforce each migrated endpoint pair with directional limits and
|
||||
lifecycle/race/failure qualification;
|
||||
6. remove legacy paths and document backend requirements.
|
||||
|
||||
Every commit must leave deterministic tests green. The first boundary
|
||||
keeps legacy endpoints operational while independently qualifying the
|
||||
new protocol. Subsequent migrations move both ends together; no commit
|
||||
requires a raw-dictionary bridge into the strict schema.
|
||||
|
||||
## Rollback Strategy
|
||||
|
||||
Revert a backend port with its paired EMS changes. Once enforcement
|
||||
begins, include the matching decorator and receive limits in that
|
||||
rollback; never leave one actor side strict while restoring a
|
||||
dictionary sender.
|
||||
|
||||
No long-lived persisted wire data requires a schema migration. Accounting and
|
||||
ledger files remain outside this protocol change. If a strictness regression
|
||||
is discovered, repair the producer or roll back the complete endpoint pair.
|
||||
Do not widen the schema to accept arbitrary dictionaries or unknown fields.
|
||||
|
||||
## Exit Criteria
|
||||
|
||||
The formal EMS messaging upgrade is complete when:
|
||||
|
||||
- all client/EMS and EMS/backend IPC payloads are tagged structs;
|
||||
- paper, Binance, Kraken, and IB share one broker startup/request/event
|
||||
contract;
|
||||
- both Tractor context families declare broad `pld_spec` aliases and narrow
|
||||
each receive direction;
|
||||
- startup, stream, and return payload failures produce tested
|
||||
`MsgTypeError` behavior;
|
||||
- no broker status outside `pending/open/canceled/closed` crosses IPC;
|
||||
- fills and existing orders use dedicated variants;
|
||||
- no business logic reconstructs wire dictionaries into message structs;
|
||||
- ack/status/cancel races are deterministic and bounded;
|
||||
- multi-client order routing is tested and the unresolved position-broadcast
|
||||
privacy limitation is explicit;
|
||||
- default EMS tests run without network or credentials;
|
||||
- actor teardown failures are asserted rather than tolerated;
|
||||
- backend authors have one conformance suite and protocol definition to follow.
|
||||
|
||||
## Primary References
|
||||
|
||||
- `piker/clearing/_messages.py`: current EMS and broker message structs.
|
||||
- `piker/clearing/_ems.py`: broker dialog opening, translation, client command
|
||||
processing, and `_emsd_main()`.
|
||||
- `piker/clearing/_client.py`: public EMS client and startup handling.
|
||||
- `piker/clearing/_paper_engine.py`: paper reference backend.
|
||||
- `piker/brokers/binance/broker.py`: Binance request and event adapter.
|
||||
- `piker/brokers/kraken/broker.py`: Kraken request and event adapter.
|
||||
- `piker/brokers/ib/broker.py`: IB request and event adapter.
|
||||
- `piker/data/validate.py`: existing datad-side `FeedInit` backend validation
|
||||
direction to improve upon rather than duplicate.
|
||||
- `piker/ui/_remote_ctl.py`: in-tree `pld_spec` and per-direction
|
||||
`limit_plds()` precedent.
|
||||
- `piker/ui/_gaps.py`: in-tree tagged union precedent.
|
||||
- `tests/test_ems.py`: current network-dependent clearing coverage and empty
|
||||
order tests.
|
||||
- `/home/goodboy/repos/tractor/tests/msg/test_pldrx_limiting.py`: canonical
|
||||
payload limiting and `MsgTypeError` behavior.
|
||||
- `/home/goodboy/repos/tractor/tests/msg/test_ext_types_msgspec.py`: canonical
|
||||
msgspec codec behavior.
|
||||
- `/home/goodboy/repos/tractor/tests/test_context_stream_semantics.py`:
|
||||
context stream cancellation and serialization semantics.
|
||||
|
||||
|
||||
Relevant issue/design threads:
|
||||
|
||||
- https://github.com/goodboy/tractor/issues/196
|
||||
- https://github.com/goodboy/tractor/issues/365
|
||||
- https://github.com/goodboy/tractor/issues/514
|
||||
- https://github.com/goodboy/tractor/pull/376
|
||||
- https://github.com/pikers/piker/issues/514
|
||||
- https://github.com/pikers/piker/issues/297
|
||||
- https://github.com/pikers/piker/issues/466
|
||||
|
|
@ -99,11 +99,11 @@ def test_schedule_request_uses_extended_hours_and_utc():
|
|||
)
|
||||
out = asyncio.run(Client.history_schedule(
|
||||
client, 'mnq.cme.20261218',
|
||||
datetime(2026, 9, 23, 4, tzinfo=UTC),
|
||||
datetime(2026, 9, 23, 0, tzinfo=UTC),
|
||||
datetime(2026, 9, 24, 1, tzinfo=UTC),
|
||||
))
|
||||
assert calls == [('contract', {
|
||||
'numDays': 1, 'endDateTime': '20260924-01:00:00',
|
||||
'numDays': 2, 'endDateTime': '20260924-01:00:00',
|
||||
'useRTH': False,
|
||||
})]
|
||||
assert out['timezone'] == 'America/New_York'
|
||||
|
|
@ -173,191 +173,3 @@ def test_schedule_duration_counts_elapsed_time_across_dst():
|
|||
datetime(2026, 11, 1, tzinfo=zone),
|
||||
datetime(2026, 11, 2, tzinfo=zone),
|
||||
))
|
||||
|
||||
|
||||
def mnq_schedule(wider: bool = False) -> SimpleNamespace:
|
||||
'''
|
||||
Reproduce the dated US/Central sessions captured from live IB.
|
||||
|
||||
'''
|
||||
starts: list[str] = ['20260923-17:00:00']
|
||||
ends: list[str] = ['20260924-16:00:00']
|
||||
if wider:
|
||||
starts.insert(0, '20260922-17:00:00')
|
||||
ends.insert(0, '20260923-16:00:00')
|
||||
sessions: list[SimpleNamespace] = []
|
||||
start: str
|
||||
end: str
|
||||
for start, end in zip(starts, ends):
|
||||
sessions.append(SimpleNamespace(
|
||||
startDateTime=start,
|
||||
endDateTime=end,
|
||||
))
|
||||
return SimpleNamespace(
|
||||
timeZone='US/Central',
|
||||
startDateTime=starts[0],
|
||||
endDateTime=ends[-1],
|
||||
sessions=sessions,
|
||||
)
|
||||
|
||||
|
||||
def test_schedule_retry_reserves_mnq_gap_outside_closure():
|
||||
'''
|
||||
One elapsed day returned only MNQ's newest trading session.
|
||||
|
||||
Its coverage began after the left gap margin, so the allocator
|
||||
could not reserve any slots. Replay both captured IB responses
|
||||
through Client.history_schedule() and reserve_history_gaps().
|
||||
Assert that widening keeps the end fixed, reserves both trading
|
||||
portions, and leaves the one-hour maintenance closure compressed.
|
||||
This uses fake transport and arrays, with no live SHM or storage.
|
||||
|
||||
'''
|
||||
import numpy as np
|
||||
from piker.tsp._reserve import reserve_history_gaps
|
||||
|
||||
calls: list[dict] = []
|
||||
|
||||
async def find_contracts(fqme: str) -> list[str]:
|
||||
return ['contract']
|
||||
|
||||
async def schedule(contract: str, **kwargs) -> SimpleNamespace:
|
||||
calls.append(kwargs)
|
||||
return mnq_schedule(wider=len(calls) > 1)
|
||||
|
||||
client = SimpleNamespace(
|
||||
find_contracts=find_contracts,
|
||||
ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
|
||||
)
|
||||
|
||||
async def query(left: float, right: float) -> dict:
|
||||
return await Client.history_schedule(
|
||||
client, 'mnq.cme.20261218',
|
||||
datetime.fromtimestamp(left, UTC),
|
||||
datetime.fromtimestamp(right, UTC),
|
||||
)
|
||||
|
||||
frame: np.ndarray = np.zeros(2, dtype=[
|
||||
('time', 'f8'),
|
||||
('open', 'f8'),
|
||||
('high', 'f8'),
|
||||
('low', 'f8'),
|
||||
('close', 'f8'),
|
||||
])
|
||||
frame['time'] = [1790193489, 1790248914]
|
||||
frame['close'] = [30755.75, 30478.5]
|
||||
result: np.ndarray
|
||||
ranges: list[tuple[float, float]]
|
||||
result, ranges = asyncio.run(reserve_history_gaps(
|
||||
frame, 1, query, max_rows=51826,
|
||||
))
|
||||
assert calls == [
|
||||
{
|
||||
'numDays': days,
|
||||
'endDateTime': '20260924-11:21:54',
|
||||
'useRTH': False,
|
||||
}
|
||||
for days in (1, 2)
|
||||
]
|
||||
assert len(result) == 51826
|
||||
assert ranges == [
|
||||
(1790193490, 1790197200),
|
||||
(1790200800, 1790248914),
|
||||
]
|
||||
assert not np.any(
|
||||
(result['time'] >= 1790197200)
|
||||
& (result['time'] < 1790200800)
|
||||
)
|
||||
np.testing.assert_array_equal(result[[0, -1]], frame)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('missing_right', [False, True])
|
||||
def test_schedule_retry_preserves_unknown_coverage(
|
||||
missing_right: bool,
|
||||
):
|
||||
'''
|
||||
Widening must not turn a partial reply into claimed coverage.
|
||||
|
||||
Return the same incomplete schedule on every fake request.
|
||||
Missing left coverage exhausts the bounded lookback; missing
|
||||
right coverage stops immediately because an earlier start cannot
|
||||
repair it. The unchanged response retains the unknown interval
|
||||
so downstream reservation still rejects it.
|
||||
|
||||
'''
|
||||
calls: list[int] = []
|
||||
|
||||
async def find_contracts(fqme: str) -> list[str]:
|
||||
return ['contract']
|
||||
|
||||
async def schedule(contract: str, **kwargs) -> SimpleNamespace:
|
||||
calls.append(kwargs['numDays'])
|
||||
return mnq_schedule()
|
||||
|
||||
client = SimpleNamespace(
|
||||
find_contracts=find_contracts,
|
||||
ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
|
||||
)
|
||||
end: datetime = datetime(2026, 9, 24, 11, tzinfo=UTC)
|
||||
if missing_right:
|
||||
end = datetime(2026, 9, 24, 23, tzinfo=UTC)
|
||||
out: dict = asyncio.run(Client.history_schedule(
|
||||
client, 'mnq.cme.20261218',
|
||||
datetime(2026, 9, 23, 19, tzinfo=UTC), end,
|
||||
))
|
||||
assert out == normalize_schedule(mnq_schedule(), False)
|
||||
assert calls == ([2] if missing_right else [1, 2, 4, 8, 16, 31])
|
||||
|
||||
|
||||
def test_schedule_retry_shares_timeout(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
'''
|
||||
A retry must not renew the caller's total schedule wait budget.
|
||||
|
||||
Record timeout scopes while the first fake reply lacks coverage
|
||||
and the second request waits forever. The shared deadline must
|
||||
cancel that second request and propagate TimeoutError, with no
|
||||
partial result substituted for failure. No sleep ordering is
|
||||
required: only the blocked second request yields to the deadline.
|
||||
|
||||
'''
|
||||
timeouts: list[float] = []
|
||||
calls: int = 0
|
||||
cancelled: bool = False
|
||||
original_timeout = asyncio.timeout
|
||||
|
||||
def timeout(delay: float):
|
||||
timeouts.append(delay)
|
||||
return original_timeout(delay)
|
||||
|
||||
monkeypatch.setattr(asyncio, 'timeout', timeout)
|
||||
|
||||
async def find_contracts(fqme: str) -> list[str]:
|
||||
return ['contract']
|
||||
|
||||
async def schedule(contract: str, **kwargs) -> SimpleNamespace:
|
||||
nonlocal calls, cancelled
|
||||
calls += 1
|
||||
if calls == 1:
|
||||
return mnq_schedule()
|
||||
try:
|
||||
await asyncio.Future()
|
||||
except asyncio.CancelledError:
|
||||
cancelled = True
|
||||
raise
|
||||
|
||||
client = SimpleNamespace(
|
||||
find_contracts=find_contracts,
|
||||
ib=SimpleNamespace(reqHistoricalScheduleAsync=schedule),
|
||||
)
|
||||
with pytest.raises(TimeoutError):
|
||||
asyncio.run(Client.history_schedule(
|
||||
client, 'mnq.cme.20261218',
|
||||
datetime(2026, 9, 23, 19, tzinfo=UTC),
|
||||
datetime(2026, 9, 24, 11, tzinfo=UTC),
|
||||
timeout_s=0.01,
|
||||
))
|
||||
assert timeouts == [0.01]
|
||||
assert calls == 2
|
||||
assert cancelled
|
||||
|
|
|
|||
|
|
@ -1,214 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Offline lifecycle conformance for every future brokerd adapter.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from piker.clearing._codec import ProtocolError
|
||||
from piker.clearing._protocol import BrokerdPhase
|
||||
from piker.clearing._schema import (
|
||||
BrokerdOpenOrder,
|
||||
BrokerdOpenOrdersComplete,
|
||||
BrokerdOrder,
|
||||
BrokerdPosition,
|
||||
BrokerdStartup,
|
||||
EmsStartup,
|
||||
Order,
|
||||
Status,
|
||||
validate_ems_startup,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('broker', ['ib', 'kraken', 'binance'])
|
||||
@pytest.mark.parametrize('mode', ['live', 'paper'])
|
||||
def test_startup_barrier(broker: str, mode: str) -> None:
|
||||
'''
|
||||
Every provider and paper mode shares one lifecycle contract.
|
||||
Prohibit order submission before completion, allow an empty
|
||||
snapshot, then reject a repeated completion or existing order.
|
||||
These fixtures qualify the protocol, not the unmigrated adapters.
|
||||
|
||||
'''
|
||||
account: str = 'paper' if mode == 'paper' else f'{broker}.test'
|
||||
phase: BrokerdPhase = BrokerdPhase(
|
||||
BrokerdStartup(
|
||||
broker=broker,
|
||||
mode=mode,
|
||||
accounts=(account,),
|
||||
)
|
||||
)
|
||||
request: BrokerdOrder = BrokerdOrder(
|
||||
oid='one',
|
||||
account=account,
|
||||
symbol='btc.spot',
|
||||
time_ns=1,
|
||||
price=Decimal('10'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
)
|
||||
with pytest.raises(ProtocolError, match='before startup'):
|
||||
phase.request(request)
|
||||
complete: BrokerdOpenOrdersComplete = BrokerdOpenOrdersComplete(
|
||||
broker=broker
|
||||
)
|
||||
assert phase.receive(complete) == complete
|
||||
assert phase.request(request) == request
|
||||
with pytest.raises(ProtocolError, match='Duplicate'):
|
||||
phase.receive(complete)
|
||||
with pytest.raises(ProtocolError, match='after startup'):
|
||||
phase.receive(
|
||||
BrokerdOpenOrder(
|
||||
reqid=1,
|
||||
time_ns=1,
|
||||
broker=broker,
|
||||
order=Order(
|
||||
oid='existing',
|
||||
account=account,
|
||||
symbol=f'btc.spot.{broker}',
|
||||
price=Decimal('10'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_paper_fallback_never_opens_live_stream() -> None:
|
||||
'''
|
||||
A backend without credentials hands control to a paper endpoint.
|
||||
Its fallback startup cannot masquerade as an empty live snapshot:
|
||||
even a syntactically valid completion event must be rejected.
|
||||
|
||||
'''
|
||||
phase: BrokerdPhase = BrokerdPhase(
|
||||
BrokerdStartup(
|
||||
broker='ib',
|
||||
mode='paper_fallback',
|
||||
)
|
||||
)
|
||||
with pytest.raises(ProtocolError, match='Fallback'):
|
||||
phase.receive(BrokerdOpenOrdersComplete(broker='ib'))
|
||||
assert not phase.complete
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'changes',
|
||||
[
|
||||
{'broker': 'kraken'},
|
||||
{'account': 'other'},
|
||||
{'symbol': 'btc.spot.kraken'},
|
||||
],
|
||||
)
|
||||
def test_event_ownership(changes: dict[str, str]) -> None:
|
||||
'''
|
||||
A well-typed event may still belong to another session. Announce
|
||||
one broker/account, then alter a position's owner or market
|
||||
suffix. ``BrokerdPhase.receive()`` must reject it before EMS can
|
||||
publish the position.
|
||||
|
||||
'''
|
||||
phase: BrokerdPhase = BrokerdPhase(
|
||||
BrokerdStartup(
|
||||
broker='ib',
|
||||
mode='live',
|
||||
accounts=('ib.test',),
|
||||
)
|
||||
)
|
||||
position: BrokerdPosition = BrokerdPosition(
|
||||
broker='ib',
|
||||
account='ib.test',
|
||||
symbol='btc.spot.ib',
|
||||
size=1.0,
|
||||
avg_price=10.0,
|
||||
)
|
||||
with pytest.raises(ProtocolError):
|
||||
phase.receive(position.copy(update=changes))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'field,value',
|
||||
[
|
||||
('oid', 'mismatch'),
|
||||
('resp', 'closed'),
|
||||
('resp', 'canceled'),
|
||||
],
|
||||
)
|
||||
def test_ems_snapshot_has_only_active_owned_dialogs(
|
||||
field: str,
|
||||
value: str,
|
||||
) -> None:
|
||||
'''
|
||||
Reconnect snapshots must not revive completed orders or associate
|
||||
a request with another oid. Corrupt one field in a valid snapshot
|
||||
and require semantic rejection after ordinary schema decoding.
|
||||
|
||||
'''
|
||||
order: Order = Order(
|
||||
oid='one',
|
||||
account='paper',
|
||||
symbol='btc.spot.ib',
|
||||
price=Decimal('10'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
)
|
||||
status: Status = Status(
|
||||
time_ns=1, oid='one', resp='open', req=order
|
||||
)
|
||||
startup: EmsStartup = EmsStartup(
|
||||
broker='ib',
|
||||
accounts=('paper',),
|
||||
positions=(),
|
||||
dialogs=(status,),
|
||||
)
|
||||
assert validate_ems_startup(startup) == startup
|
||||
bad: EmsStartup = startup.copy(
|
||||
update={
|
||||
'dialogs': (status.copy(update={field: value}),),
|
||||
}
|
||||
)
|
||||
with pytest.raises(ProtocolError):
|
||||
validate_ems_startup(bad)
|
||||
|
||||
|
||||
def test_ems_startup_requires_an_account() -> None:
|
||||
'''
|
||||
EMS readiness means a live or paper account can accept requests.
|
||||
An empty snapshot is legal, but an empty account set cannot
|
||||
provide that service. Reject it even with no positions or
|
||||
dialogs.
|
||||
|
||||
'''
|
||||
with pytest.raises(
|
||||
ProtocolError,
|
||||
match='requires an account',
|
||||
):
|
||||
validate_ems_startup(
|
||||
EmsStartup(
|
||||
broker='ib',
|
||||
accounts=(),
|
||||
positions=(),
|
||||
dialogs=(),
|
||||
)
|
||||
)
|
||||
|
|
@ -1,458 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Real local-actor qualification of clearing's directional contracts.
|
||||
|
||||
No market-data service, credentials, or persistent account is opened.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
import tractor
|
||||
from tractor.trionics import collapse_eg
|
||||
import trio
|
||||
|
||||
from piker.clearing._codec import ProtocolError, validate
|
||||
from piker.clearing._protocol import (
|
||||
BrokerdPhase,
|
||||
Stream,
|
||||
open_stream,
|
||||
)
|
||||
from piker.clearing._schema import (
|
||||
BrokerdEvent,
|
||||
BrokerdFill,
|
||||
BrokerdOpenOrder,
|
||||
BrokerdOpenOrdersComplete,
|
||||
BrokerdOrder,
|
||||
BrokerdOrderAck,
|
||||
BrokerdPayload,
|
||||
BrokerdRequest,
|
||||
BrokerdStartup,
|
||||
Order,
|
||||
validate_startup,
|
||||
BrokerdStatus,
|
||||
Cancel,
|
||||
EmsEvent,
|
||||
EmsPayload,
|
||||
EmsRequest,
|
||||
EmsStartup,
|
||||
Status,
|
||||
validate_ems_startup,
|
||||
)
|
||||
from piker.clearing._state import Effects, OrderState
|
||||
|
||||
|
||||
@tractor.context(pld_spec=BrokerdPayload)
|
||||
async def fake_broker(
|
||||
ctx: tractor.Context,
|
||||
fault: str,
|
||||
empty: bool,
|
||||
) -> None:
|
||||
'''
|
||||
Exchange typed broker traffic or inject a selected wire fault.
|
||||
|
||||
``fault`` selects a provider exception, wrong direction, wrong
|
||||
phase, or malformed execution. ``empty`` omits the existing order
|
||||
while retaining the explicit completion barrier.
|
||||
|
||||
'''
|
||||
startup: BrokerdStartup = BrokerdStartup(
|
||||
broker='fixture',
|
||||
mode='paper',
|
||||
accounts=('paper',),
|
||||
)
|
||||
await ctx.started(validate_startup(startup))
|
||||
stream: Stream[BrokerdRequest, BrokerdEvent]
|
||||
async with open_stream(
|
||||
ctx,
|
||||
BrokerdRequest,
|
||||
BrokerdEvent,
|
||||
) as stream:
|
||||
if not empty:
|
||||
await stream.send(
|
||||
BrokerdOpenOrder(
|
||||
broker='fixture',
|
||||
reqid='existing-1',
|
||||
time_ns=1,
|
||||
order=Order(
|
||||
oid='fixture:paper:existing-1',
|
||||
account='paper',
|
||||
symbol='btc.spot.fixture',
|
||||
price=Decimal('12.25'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
),
|
||||
)
|
||||
)
|
||||
await stream.send(
|
||||
BrokerdOpenOrdersComplete(
|
||||
broker='fixture',
|
||||
)
|
||||
)
|
||||
request: BrokerdOrder = await stream.receive()
|
||||
if fault == 'peer_error':
|
||||
raise RuntimeError('fixture provider failed')
|
||||
if fault == 'wrong_direction':
|
||||
# Deliberately bypass the local send guard to prove that
|
||||
# the other actor's scoped wire decoder rejects this.
|
||||
await stream.raw.send(request)
|
||||
elif fault == 'wrong_phase':
|
||||
await stream.raw.send(startup)
|
||||
elif fault == 'malformed':
|
||||
await stream.raw.send(
|
||||
{
|
||||
'type': 'broker_fill',
|
||||
'time_ns': 1,
|
||||
'broker_time': 1.0,
|
||||
'reqid': 1,
|
||||
'account': 'paper',
|
||||
'broker': 'fixture',
|
||||
'action': 'buy',
|
||||
'size': 1.0,
|
||||
'price': 12.25,
|
||||
}
|
||||
)
|
||||
else:
|
||||
await stream.send(
|
||||
BrokerdOrderAck(
|
||||
oid=request.oid,
|
||||
reqid=1,
|
||||
account='paper',
|
||||
)
|
||||
)
|
||||
await stream.send(
|
||||
BrokerdFill(
|
||||
time_ns=1,
|
||||
broker_time=1.0,
|
||||
reqid=1,
|
||||
account='paper',
|
||||
broker='fixture',
|
||||
fill_id='fill-1',
|
||||
action='buy',
|
||||
size=1.0,
|
||||
price=12.25,
|
||||
)
|
||||
)
|
||||
# A malicious peer must not hold teardown open after its
|
||||
# receiver rejects a payload. The parent cancels this sleep.
|
||||
await trio.sleep_forever()
|
||||
|
||||
|
||||
async def exercise(fault: str = '', empty: bool = False) -> None:
|
||||
'''
|
||||
Own the fake broker actor tree through exchange and teardown.
|
||||
|
||||
``fault`` and ``empty`` select ``fake_broker`` behavior. A Trio
|
||||
deadline encloses all actor scopes; the test runner adds an
|
||||
independent process timeout.
|
||||
|
||||
'''
|
||||
# The deadline encloses the entire actor tree, including
|
||||
# teardown.
|
||||
# An outer process timeout remains necessary for shielded runtime
|
||||
# failures; the run-tests harness supplies that second bound.
|
||||
with trio.fail_after(15):
|
||||
nursery: tractor.ActorNursery
|
||||
async with (
|
||||
collapse_eg(),
|
||||
tractor.open_root_actor(
|
||||
name='clearing-schema-test',
|
||||
tpt_bind_addrs=[('127.0.0.1', 0)],
|
||||
),
|
||||
tractor.open_nursery() as nursery,
|
||||
):
|
||||
portal: tractor.Portal = await nursery.start_actor(
|
||||
'fake-broker',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
ctx: tractor.Context
|
||||
startup: BrokerdStartup
|
||||
async with portal.open_context(
|
||||
fake_broker,
|
||||
fault=fault,
|
||||
empty=empty,
|
||||
) as (ctx, startup):
|
||||
phase: BrokerdPhase = BrokerdPhase(startup)
|
||||
stream: Stream[BrokerdEvent, BrokerdRequest]
|
||||
async with open_stream(
|
||||
ctx,
|
||||
BrokerdEvent,
|
||||
BrokerdRequest,
|
||||
) as stream:
|
||||
if not empty:
|
||||
existing: BrokerdEvent = phase.receive(
|
||||
await stream.receive(),
|
||||
)
|
||||
assert isinstance(existing, BrokerdOpenOrder)
|
||||
assert isinstance(existing.order, Order)
|
||||
assert existing.order.price == Decimal(
|
||||
'12.25',
|
||||
)
|
||||
phase.receive(await stream.receive())
|
||||
assert phase.complete
|
||||
request: BrokerdOrder = BrokerdOrder(
|
||||
oid='client-1',
|
||||
account='paper',
|
||||
time_ns=1,
|
||||
symbol='btc.spot',
|
||||
price=Decimal('12.25'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
)
|
||||
await stream.send(phase.request(request))
|
||||
ack: BrokerdEvent = phase.receive(
|
||||
await stream.receive()
|
||||
)
|
||||
assert type(ack) is BrokerdOrderAck
|
||||
fill: BrokerdEvent = phase.receive(
|
||||
await stream.receive()
|
||||
)
|
||||
assert type(fill) is BrokerdFill
|
||||
assert fill.fill_id == 'fill-1'
|
||||
await ctx.cancel()
|
||||
await nursery.cancel()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('empty', [True, False])
|
||||
def test_typed_startup_and_stream(empty: bool) -> None:
|
||||
'''
|
||||
Typed startup must precede an explicit completion barrier, even
|
||||
with no existing orders. Exercise both snapshots through a real
|
||||
child process and exchange a request, ack and identified fill.
|
||||
Nested Order and Decimal types must survive actor serialization.
|
||||
|
||||
'''
|
||||
trio.run(exercise, '', empty)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'fault',
|
||||
[
|
||||
'wrong_direction',
|
||||
'wrong_phase',
|
||||
'malformed',
|
||||
],
|
||||
)
|
||||
def test_wire_rejection_and_bounded_teardown(fault: str) -> None:
|
||||
'''
|
||||
A peer bypassing outbound checks must still fail at wire decode.
|
||||
Send a request in the event direction, a startup in the stream,
|
||||
or a fill without its execution ID. Require MsgTypeError and let
|
||||
exercise's deadline cover cancellation of the sleeping producer.
|
||||
|
||||
'''
|
||||
with pytest.raises(tractor.MsgTypeError):
|
||||
trio.run(exercise, fault)
|
||||
|
||||
|
||||
def test_provider_failure_is_not_swallowed() -> None:
|
||||
'''
|
||||
Provider failure while a client waits for an ack must abort that
|
||||
receive and terminate the actor tree. Raise inside the fake
|
||||
broker and require its remote RuntimeError within the teardown
|
||||
deadline.
|
||||
|
||||
'''
|
||||
error: pytest.ExceptionInfo[tractor.RemoteActorError]
|
||||
with pytest.raises(tractor.RemoteActorError) as error:
|
||||
trio.run(exercise, 'peer_error')
|
||||
assert error.value.boxed_type is RuntimeError
|
||||
|
||||
|
||||
@tractor.context(pld_spec=BrokerdPayload)
|
||||
async def wrong_final_result(
|
||||
ctx: tractor.Context,
|
||||
) -> BrokerdStartup:
|
||||
'''
|
||||
Return a wire-valid startup in the forbidden final phase.
|
||||
|
||||
'''
|
||||
startup: BrokerdStartup = BrokerdStartup(
|
||||
broker='fixture',
|
||||
mode='paper',
|
||||
accounts=('paper',),
|
||||
)
|
||||
await ctx.started(startup)
|
||||
return startup
|
||||
|
||||
|
||||
def test_final_result_has_its_own_phase() -> None:
|
||||
'''
|
||||
A lifecycle union permits startup objects but a finished broker
|
||||
context returns only None. Receive a wire-valid wrong Return and
|
||||
prove the phase validator rejects it after the child terminates.
|
||||
|
||||
'''
|
||||
|
||||
async def main() -> None:
|
||||
'''
|
||||
Receive a completed child result and reject its startup
|
||||
payload.
|
||||
|
||||
'''
|
||||
with trio.fail_after(15):
|
||||
nursery: tractor.ActorNursery
|
||||
async with (
|
||||
collapse_eg(),
|
||||
tractor.open_root_actor(
|
||||
tpt_bind_addrs=[('127.0.0.1', 0)],
|
||||
),
|
||||
tractor.open_nursery() as nursery,
|
||||
):
|
||||
portal: tractor.Portal = await nursery.start_actor(
|
||||
'wrong-return',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
ctx: tractor.Context
|
||||
_: BrokerdStartup
|
||||
async with portal.open_context(
|
||||
wrong_final_result,
|
||||
) as (ctx, _):
|
||||
result: BrokerdPayload = (
|
||||
await ctx.wait_for_result()
|
||||
)
|
||||
with pytest.raises(ProtocolError):
|
||||
validate(result, type(None))
|
||||
await nursery.cancel()
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
@tractor.context(pld_spec=EmsPayload)
|
||||
async def fake_ems(ctx: tractor.Context) -> None:
|
||||
'''
|
||||
Apply client cancellation before publishing the provider ack.
|
||||
|
||||
The receive order fixes the interleaving: ``OrderState.cancel()``
|
||||
runs before ``OrderState.accept()`` handles the ack. Assert one
|
||||
deferred cancellation and publish its terminal response.
|
||||
|
||||
'''
|
||||
await ctx.started(
|
||||
EmsStartup(
|
||||
broker='fixture',
|
||||
positions=(),
|
||||
accounts=('paper',),
|
||||
dialogs=(),
|
||||
)
|
||||
)
|
||||
state: OrderState = OrderState()
|
||||
stream: Stream[EmsRequest, EmsEvent]
|
||||
async with open_stream(ctx, EmsRequest, EmsEvent) as stream:
|
||||
order: Order = await stream.receive()
|
||||
state.register(order)
|
||||
cancel: Cancel = await stream.receive()
|
||||
assert cancel.oid == order.oid
|
||||
assert not state.cancel(cancel.oid).broker
|
||||
effects: Effects = state.accept(
|
||||
BrokerdOrderAck(
|
||||
oid=order.oid,
|
||||
account=order.account,
|
||||
reqid=1,
|
||||
)
|
||||
)
|
||||
assert len(effects.broker) == 1
|
||||
assert effects.broker[0].oid == order.oid
|
||||
effects = state.accept(
|
||||
BrokerdStatus(
|
||||
time_ns=1,
|
||||
account=order.account,
|
||||
broker='fixture',
|
||||
reqid=1,
|
||||
status='canceled',
|
||||
)
|
||||
)
|
||||
await stream.send(effects.clients[0])
|
||||
await trio.sleep_forever()
|
||||
|
||||
|
||||
def test_ems_cancel_before_ack_over_ipc() -> None:
|
||||
'''
|
||||
Client Order and Cancel must arrive as typed requests and produce
|
||||
an immutable nested Status. Hold the fake provider ack until the
|
||||
cancel reaches the remote EMS reducer, then require one deferred
|
||||
broker cancel and a terminal client response through real IPC.
|
||||
|
||||
'''
|
||||
|
||||
async def main() -> None:
|
||||
'''
|
||||
Send ordered client intent and require a typed canceled
|
||||
snapshot.
|
||||
|
||||
'''
|
||||
with trio.fail_after(15):
|
||||
nursery: tractor.ActorNursery
|
||||
async with (
|
||||
collapse_eg(),
|
||||
tractor.open_root_actor(
|
||||
tpt_bind_addrs=[('127.0.0.1', 0)],
|
||||
),
|
||||
tractor.open_nursery() as nursery,
|
||||
):
|
||||
portal: tractor.Portal = await nursery.start_actor(
|
||||
'fake-ems',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
ctx: tractor.Context
|
||||
startup: EmsStartup
|
||||
async with portal.open_context(
|
||||
fake_ems,
|
||||
) as (ctx, startup):
|
||||
startup = validate_ems_startup(
|
||||
startup
|
||||
)
|
||||
assert startup.dialogs == ()
|
||||
stream: Stream[EmsEvent, EmsRequest]
|
||||
async with open_stream(
|
||||
ctx,
|
||||
EmsEvent,
|
||||
EmsRequest,
|
||||
) as stream:
|
||||
order: Order = Order(
|
||||
oid='client-1',
|
||||
account='paper',
|
||||
symbol='btc.spot.fixture',
|
||||
price=Decimal('12.25'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
)
|
||||
await stream.send(order)
|
||||
await stream.send(
|
||||
Cancel(
|
||||
oid=order.oid,
|
||||
symbol=order.symbol,
|
||||
)
|
||||
)
|
||||
status: EmsEvent = await stream.receive()
|
||||
assert isinstance(status, Status)
|
||||
assert status.req == order
|
||||
assert status.resp == 'canceled'
|
||||
assert status.cancel_called
|
||||
assert isinstance(
|
||||
status.brokerd_event,
|
||||
BrokerdStatus,
|
||||
)
|
||||
await ctx.cancel()
|
||||
await nursery.cancel()
|
||||
|
||||
trio.run(main)
|
||||
|
|
@ -1,333 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Offline clearing schema and local API boundary conformance.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
from types import UnionType
|
||||
|
||||
import msgspec
|
||||
from msgspec import msgpack
|
||||
import pytest
|
||||
|
||||
from piker.clearing._codec import (
|
||||
Message,
|
||||
ProtocolError,
|
||||
decoder,
|
||||
validate,
|
||||
)
|
||||
from piker.clearing._schema import (
|
||||
BrokerdCancel,
|
||||
BrokerdError,
|
||||
BrokerdEvent,
|
||||
BrokerdFill,
|
||||
BrokerdOpenOrder,
|
||||
BrokerdOpenOrdersComplete,
|
||||
BrokerdOrder,
|
||||
BrokerdOrderAck,
|
||||
BrokerdPayload,
|
||||
BrokerdPosition,
|
||||
BrokerdRequest,
|
||||
BrokerdStartup,
|
||||
BrokerdStatus,
|
||||
Cancel,
|
||||
EmsEvent,
|
||||
EmsPayload,
|
||||
EmsRequest,
|
||||
EmsStartup,
|
||||
Order,
|
||||
Status,
|
||||
validate_startup,
|
||||
)
|
||||
|
||||
|
||||
ORDER: Order = Order(
|
||||
oid='client-1',
|
||||
symbol='btc.usdt.binance',
|
||||
account='paper',
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
price=Decimal('12.25'),
|
||||
size=1.0,
|
||||
brokers=('binance',),
|
||||
)
|
||||
REQUEST: BrokerdOrder = BrokerdOrder(
|
||||
oid=ORDER.oid,
|
||||
account=ORDER.account,
|
||||
symbol='btc.usdt',
|
||||
time_ns=1,
|
||||
price=ORDER.price,
|
||||
size=ORDER.size,
|
||||
action='buy',
|
||||
)
|
||||
POSITION: BrokerdPosition = BrokerdPosition(
|
||||
broker='binance',
|
||||
account='paper',
|
||||
symbol=ORDER.symbol,
|
||||
size=1.0,
|
||||
avg_price=12.25,
|
||||
)
|
||||
EVENTS: tuple[BrokerdEvent, ...] = (
|
||||
BrokerdOrderAck(oid=ORDER.oid, reqid=7, account='paper'),
|
||||
BrokerdStatus(
|
||||
time_ns=1,
|
||||
reqid=7,
|
||||
status='open',
|
||||
account='paper',
|
||||
broker='binance',
|
||||
),
|
||||
BrokerdFill(
|
||||
time_ns=1,
|
||||
broker_time=1.0,
|
||||
reqid=7,
|
||||
fill_id='exec-1',
|
||||
account='paper',
|
||||
broker='binance',
|
||||
action='buy',
|
||||
size=1.0,
|
||||
price=12.25,
|
||||
),
|
||||
BrokerdError(
|
||||
account='paper',
|
||||
broker='binance',
|
||||
reason='rejected',
|
||||
request=REQUEST,
|
||||
),
|
||||
POSITION,
|
||||
BrokerdOpenOrder(
|
||||
reqid=7,
|
||||
order=ORDER,
|
||||
broker='binance',
|
||||
time_ns=1,
|
||||
),
|
||||
BrokerdOpenOrdersComplete(broker='binance'),
|
||||
)
|
||||
STATUS: Status = Status(
|
||||
time_ns=1,
|
||||
oid=ORDER.oid,
|
||||
resp='fill',
|
||||
req=ORDER,
|
||||
brokerd_event=EVENTS[2],
|
||||
)
|
||||
STARTUP: BrokerdStartup = BrokerdStartup(
|
||||
broker='binance',
|
||||
mode='paper',
|
||||
accounts=('paper',),
|
||||
positions=(POSITION,),
|
||||
)
|
||||
|
||||
|
||||
event: BrokerdEvent
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'msg,spec',
|
||||
[
|
||||
(ORDER, EmsRequest),
|
||||
(Cancel(oid=ORDER.oid, symbol=ORDER.symbol), EmsRequest),
|
||||
(REQUEST, BrokerdRequest),
|
||||
(
|
||||
BrokerdCancel(
|
||||
oid=ORDER.oid,
|
||||
symbol=ORDER.symbol,
|
||||
account='paper',
|
||||
time_ns=1,
|
||||
reqid=7,
|
||||
),
|
||||
BrokerdRequest,
|
||||
),
|
||||
*((event, BrokerdEvent) for event in EVENTS),
|
||||
(STATUS, EmsEvent),
|
||||
(STARTUP, BrokerdPayload),
|
||||
(
|
||||
EmsStartup(
|
||||
broker='binance',
|
||||
positions=(POSITION,),
|
||||
accounts=('paper',),
|
||||
dialogs=(STATUS,),
|
||||
),
|
||||
EmsPayload,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_protocol_roundtrip(
|
||||
msg: Message|None,
|
||||
spec: type|UnionType,
|
||||
) -> None:
|
||||
'''
|
||||
Every clearing variant must round-trip through its real union,
|
||||
including nested orders, fills, Decimal prices and startup
|
||||
tuples. Decoding with the declared peer spec proves
|
||||
discriminators are unique and fields are supported without hooks
|
||||
or Any fallbacks.
|
||||
|
||||
'''
|
||||
decoded: Message = decoder(spec).decode(msgpack.encode(msg))
|
||||
assert type(decoded) is type(msg)
|
||||
assert decoded == msg
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'msg,spec',
|
||||
[
|
||||
(ORDER, BrokerdRequest),
|
||||
(REQUEST, BrokerdEvent),
|
||||
(EVENTS[0], BrokerdRequest),
|
||||
(STARTUP, BrokerdEvent),
|
||||
(STATUS, EmsRequest),
|
||||
(None, EmsRequest),
|
||||
],
|
||||
)
|
||||
def test_wrong_direction_and_phase(
|
||||
msg: Message|None,
|
||||
spec: type|UnionType,
|
||||
) -> None:
|
||||
'''
|
||||
A valid lifecycle message is invalid in another direction or
|
||||
streaming phase. Test both the local send guard and wire decoder:
|
||||
sharing a transport does not widen an endpoint's receive set.
|
||||
|
||||
'''
|
||||
with pytest.raises(ProtocolError):
|
||||
validate(msg, spec)
|
||||
with pytest.raises(msgspec.ValidationError):
|
||||
decoder(spec).decode(msgpack.encode(msg))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'changes',
|
||||
[
|
||||
{'extra': 'unknown'},
|
||||
{'action': 'cancel'},
|
||||
{'size': '1.0'},
|
||||
{'account': None},
|
||||
{'price': []},
|
||||
],
|
||||
)
|
||||
def test_invalid_order_fields(changes: dict[str, object]) -> None:
|
||||
'''
|
||||
Foreign fields and incorrectly typed fields must fail at decode.
|
||||
Modify a genuine encoded order so only the tested field changes;
|
||||
successful parsing of the remaining fields cannot hide the error.
|
||||
|
||||
'''
|
||||
raw: dict[str, object] = (
|
||||
msgpack.decode(msgpack.encode(ORDER)) | changes
|
||||
)
|
||||
with pytest.raises(msgspec.ValidationError):
|
||||
decoder(EmsRequest).decode(msgpack.encode(raw))
|
||||
|
||||
|
||||
def test_construction_is_not_validation() -> None:
|
||||
'''
|
||||
msgspec constructors accept an invalid status literal. The local
|
||||
boundary must reject it before any transport send is attempted. A
|
||||
structurally valid tagged dictionary is also rejected locally,
|
||||
although its wire representation correctly decodes to a Struct.
|
||||
|
||||
'''
|
||||
bad: BrokerdStatus = BrokerdStatus(
|
||||
time_ns=1,
|
||||
reqid=7,
|
||||
status='filled',
|
||||
account='paper',
|
||||
broker='binance',
|
||||
)
|
||||
with pytest.raises(msgspec.ValidationError):
|
||||
validate(bad, BrokerdEvent)
|
||||
raw: dict[str, object] = msgpack.decode(msgpack.encode(ORDER))
|
||||
with pytest.raises(ProtocolError):
|
||||
validate(raw, EmsRequest)
|
||||
assert decoder(EmsRequest).decode(msgpack.encode(raw)) == ORDER
|
||||
|
||||
|
||||
def test_immutable_snapshot_and_copy() -> None:
|
||||
'''
|
||||
Editing an order must not mutate a previously published snapshot.
|
||||
Exercise the public copy method, frozen fields and nested request
|
||||
validation so an invalid replacement cannot enter local history.
|
||||
|
||||
'''
|
||||
replacement: Order = ORDER.copy(update={'price': Decimal('13')})
|
||||
assert ORDER.price == Decimal('12.25')
|
||||
assert replacement.price == Decimal('13')
|
||||
with pytest.raises(AttributeError):
|
||||
ORDER.price = Decimal('14')
|
||||
with pytest.raises(msgspec.ValidationError):
|
||||
STATUS.copy(
|
||||
update={
|
||||
'req': msgspec.structs.replace(ORDER, size='bad'),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'changes',
|
||||
[
|
||||
{'accounts': ('paper', 'paper')},
|
||||
{'accounts': (), 'positions': ()},
|
||||
{'mode': 'live', 'accounts': (), 'positions': ()},
|
||||
{'accounts': ('other',), 'positions': ()},
|
||||
{'accounts': ('other',)},
|
||||
{'broker': 'kraken'},
|
||||
{'positions': (POSITION, POSITION)},
|
||||
{'mode': 'paper_fallback'},
|
||||
],
|
||||
)
|
||||
def test_startup_ownership(changes: dict[str, object]) -> None:
|
||||
'''
|
||||
A typed startup can still claim another backend's positions or
|
||||
duplicate an account. Alter one invariant at a time and verify
|
||||
the semantic validator rejects it before EMS publishes the state.
|
||||
|
||||
'''
|
||||
with pytest.raises(ProtocolError):
|
||||
validate_startup(STARTUP.copy(update=changes))
|
||||
assert validate_startup(STARTUP) == STARTUP
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'bad',
|
||||
[
|
||||
{'req': msgpack.decode(msgpack.encode(ORDER))},
|
||||
{
|
||||
'req': msgspec.structs.replace(
|
||||
ORDER,
|
||||
price=Decimal('NaN'),
|
||||
),
|
||||
},
|
||||
{'req': msgspec.structs.replace(ORDER, price='12.25')},
|
||||
{'req': msgspec.structs.replace(ORDER, brokers=['binance'])},
|
||||
{'req': msgspec.structs.replace(ORDER, size=float('inf'))},
|
||||
],
|
||||
)
|
||||
def test_local_nested_objects_are_typed(
|
||||
bad: dict[str, object],
|
||||
) -> None:
|
||||
'''
|
||||
A tagged nested map can encode identically to an Order and a
|
||||
string can encode identically to Decimal. The local API must
|
||||
still require typed objects, immutable tuples and finite prices.
|
||||
Replace nested fields without constructor validation and require
|
||||
rejection before the snapshot can enter a queue or order history.
|
||||
|
||||
'''
|
||||
with pytest.raises(ProtocolError):
|
||||
validate(msgspec.structs.replace(STATUS, **bad), EmsEvent)
|
||||
|
|
@ -1,332 +0,0 @@
|
|||
# piker: trading gear for hackers
|
||||
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
||||
|
||||
# This program is free software: you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Affero General Public
|
||||
# License as published by the Free Software Foundation, either
|
||||
# version 3 of the License, or (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public
|
||||
# License along with this program. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Deterministic order-event sequencing without feeds or credentials.
|
||||
|
||||
'''
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from piker.clearing._codec import ProtocolError
|
||||
from piker.clearing._schema import (
|
||||
BrokerdCancel,
|
||||
BrokerdError,
|
||||
BrokerdFill,
|
||||
BrokerdOrderAck,
|
||||
BrokerdStatus,
|
||||
Order,
|
||||
Status,
|
||||
)
|
||||
from piker.clearing._state import Effects, OrderState
|
||||
|
||||
|
||||
def order(
|
||||
oid: str = 'one',
|
||||
account: str = 'paper',
|
||||
) -> Order:
|
||||
'''
|
||||
Build live client intent with controllable dialog/account
|
||||
identity.
|
||||
|
||||
'''
|
||||
return Order(
|
||||
oid=oid,
|
||||
account=account,
|
||||
symbol='btc.usdt.binance',
|
||||
price=Decimal('10'),
|
||||
size=1.0,
|
||||
action='buy',
|
||||
exec_mode='live',
|
||||
)
|
||||
|
||||
|
||||
def ack(
|
||||
oid: str = 'one',
|
||||
account: str = 'paper',
|
||||
reqid: int|str = 1,
|
||||
) -> BrokerdOrderAck:
|
||||
'''
|
||||
Build the acknowledgement establishing an account/provider ID.
|
||||
|
||||
'''
|
||||
return BrokerdOrderAck(oid=oid, account=account, reqid=reqid)
|
||||
|
||||
|
||||
def fill(
|
||||
fill_id: str = 'fill-1',
|
||||
account: str = 'paper',
|
||||
reqid: int|str = 1,
|
||||
) -> BrokerdFill:
|
||||
'''
|
||||
Build an execution whose delivery may precede its
|
||||
acknowledgement.
|
||||
|
||||
'''
|
||||
return BrokerdFill(
|
||||
time_ns=1,
|
||||
broker_time=1.0,
|
||||
account=account,
|
||||
broker='binance',
|
||||
reqid=reqid,
|
||||
fill_id=fill_id,
|
||||
size=0.5,
|
||||
price=10.0,
|
||||
action='buy',
|
||||
)
|
||||
|
||||
|
||||
def closed() -> BrokerdStatus:
|
||||
'''
|
||||
Build a terminal provider status for the default test order.
|
||||
|
||||
'''
|
||||
return BrokerdStatus(
|
||||
time_ns=1,
|
||||
reqid=1,
|
||||
account='paper',
|
||||
broker='binance',
|
||||
status='closed',
|
||||
)
|
||||
|
||||
|
||||
def test_cancel_before_ack_once() -> None:
|
||||
'''
|
||||
A cancel can reach EMS while broker submission is suspended.
|
||||
Register intent, cancel twice, then repeat the ack. Only the
|
||||
first ack may produce a cancel, carrying the original account and
|
||||
oid. The earlier published snapshot must remain unchanged.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
initial: Status = state.register(order())
|
||||
assert not state.cancel('one').broker
|
||||
assert not state.cancel('one').broker
|
||||
effects: Effects = state.accept(ack())
|
||||
assert len(effects.broker) == 1
|
||||
cancel: BrokerdCancel = effects.broker[0]
|
||||
assert (cancel.oid, cancel.reqid, cancel.account) == (
|
||||
'one',
|
||||
1,
|
||||
'paper',
|
||||
)
|
||||
assert not state.accept(ack()).broker
|
||||
assert initial.cancel_called is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize('terminal_first', [True, False])
|
||||
def test_early_duplicate_and_late_fills(
|
||||
terminal_first: bool,
|
||||
) -> None:
|
||||
'''
|
||||
Native callbacks may precede an ack and a final fill may follow
|
||||
closed. Buffer both event orders, correlate with the ack, then
|
||||
repeat the fill and inject a distinct late execution. Each fill
|
||||
must be emitted once; neither may reopen the completed dialog.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
state.register(order())
|
||||
events: list[BrokerdStatus|BrokerdFill] = (
|
||||
[closed(), fill()] if terminal_first else [fill(), closed()]
|
||||
)
|
||||
event: BrokerdStatus|BrokerdFill
|
||||
for event in events:
|
||||
assert not state.accept(event).clients
|
||||
effects: Effects = state.accept(ack())
|
||||
responses: list[str] = []
|
||||
msg: Status
|
||||
for msg in effects.clients:
|
||||
responses.append(msg.resp)
|
||||
assert responses == (
|
||||
['closed', 'fill'] if terminal_first else ['fill', 'closed']
|
||||
)
|
||||
assert not state.accept(fill()).clients
|
||||
assert state.accept(fill('fill-2')).clients[0].resp == 'fill'
|
||||
assert not state.active
|
||||
assert state.terminal['one'].status.resp == 'closed'
|
||||
|
||||
|
||||
def test_closed_before_ack_cancels_no_finished_order() -> None:
|
||||
'''
|
||||
An order can finish while its ack is delayed and cancel is
|
||||
queued. Replay the pre-ack terminal status before generating
|
||||
effects so the delayed acknowledgement cannot cancel an already
|
||||
closed order.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
state.register(order())
|
||||
state.cancel('one')
|
||||
state.accept(closed())
|
||||
assert not state.accept(ack()).broker
|
||||
|
||||
|
||||
def test_accounts_partition_request_ids() -> None:
|
||||
'''
|
||||
Separate broker sessions may both allocate request ID one.
|
||||
Register and acknowledge two accounts, then route their fills.
|
||||
Correlation must retain account ownership even when IDs collide.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
oid: str
|
||||
account: str
|
||||
for oid, account in [('one', 'paper'), ('two', 'other')]:
|
||||
state.register(order(oid, account))
|
||||
state.accept(ack(oid, account))
|
||||
effect: Effects = state.accept(fill(account=account))
|
||||
assert effect.clients[0].oid == oid
|
||||
|
||||
|
||||
def test_pre_ack_capacity_is_bounded() -> None:
|
||||
'''
|
||||
A backend that never acknowledges must not grow EMS memory
|
||||
forever. Exceed a deliberately tiny buffer and assert a protocol
|
||||
failure preserves the already buffered execution instead of
|
||||
dropping it.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState(max_early=1)
|
||||
state.accept(fill())
|
||||
with pytest.raises(ProtocolError, match='capacity'):
|
||||
state.accept(fill('fill-2'))
|
||||
assert state.early[('paper', 1)] == [fill()]
|
||||
|
||||
|
||||
def test_tombstone_eviction_removes_correlation() -> None:
|
||||
'''
|
||||
Terminal dialogs have finite retention. Complete two orders with
|
||||
a one-entry limit; evicting the first must also remove every old
|
||||
request ID so a later event cannot target a missing dialog.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState(max_terminal=1)
|
||||
oid: str
|
||||
reqid: int
|
||||
for oid, reqid in [('one', 1), ('two', 2)]:
|
||||
state.register(order(oid))
|
||||
state.accept(ack(oid, reqid=reqid))
|
||||
state.accept(closed().copy(update={'reqid': reqid}))
|
||||
assert tuple(state.terminal) == ('two',)
|
||||
assert state.reqids == {('paper', 2): 'two'}
|
||||
|
||||
|
||||
def test_edit_before_ack_cannot_duplicate_submission() -> None:
|
||||
'''
|
||||
An edit with no known backend request ID could place a second
|
||||
order. Hold the ack and try an edit; rejection must leave the
|
||||
original request intact for subsequent acknowledgement and
|
||||
cancellation.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
initial: Status = state.register(order())
|
||||
with pytest.raises(
|
||||
ProtocolError,
|
||||
match='before acknowledgement',
|
||||
):
|
||||
state.register(order().copy(update={'price': Decimal('11')}))
|
||||
assert state.active['one'] is initial
|
||||
|
||||
|
||||
def test_edit_ack_keeps_late_fills_on_original_dialog() -> None:
|
||||
'''
|
||||
An edit may replace the provider request ID while executions for
|
||||
the old ID remain in flight. A delayed duplicate ack must not
|
||||
revert the cancel target, and either ID must still route fills to
|
||||
the same EMS order. Cancel once per accepted request identity.
|
||||
|
||||
'''
|
||||
state: OrderState = OrderState()
|
||||
state.register(order())
|
||||
state.accept(ack())
|
||||
state.register(order().copy(update={'price': Decimal('11')}))
|
||||
state.accept(ack(reqid=2))
|
||||
state.accept(ack(reqid=1))
|
||||
assert state.active['one'].reqid == 2
|
||||
assert state.cancel('one').broker[0].reqid == 2
|
||||
assert not state.cancel('one').broker
|
||||
assert state.accept(fill(reqid=1)).clients[0].oid == 'one'
|
||||
effects: Effects = state.accept(fill('fill-2', reqid=2))
|
||||
assert effects.clients[0].oid == 'one'
|
||||
assert state.active['one'].reqid == 2
|
||||
state.accept(closed())
|
||||
assert 'one' in state.active
|
||||
|
||||
|
||||
def test_startup_collision_does_not_register_second_order() -> None:
|
||||
'''
|
||||
Two existing orders cannot claim one account/request ID pair.
|
||||
Correlate a normal order, then inject a startup order with the
|
||||
same provider ID. Rejection leaves ``OrderState.active`` and
|
||||
``OrderState.dialogs`` free of the rejected second order.
|
||||
|
||||
'''
|
||||
from piker.clearing._schema import BrokerdOpenOrder
|
||||
|
||||
state: OrderState = OrderState()
|
||||
state.register(order())
|
||||
state.accept(ack())
|
||||
with pytest.raises(ProtocolError, match='ownership'):
|
||||
state.accept(
|
||||
BrokerdOpenOrder(
|
||||
reqid=1,
|
||||
order=order('two'),
|
||||
broker='binance',
|
||||
time_ns=1,
|
||||
)
|
||||
)
|
||||
assert tuple(state.dialogs) == ('one',)
|
||||
assert tuple(state.active) == ('one',)
|
||||
|
||||
|
||||
def test_submission_and_edit_rejections() -> None:
|
||||
'''
|
||||
A new submission error has no future ack to unblock cancellation.
|
||||
Its typed request identifies a terminal rejection; an edit error
|
||||
must retain the original live control ID so it can still cancel.
|
||||
Exercise both with identical reasons to prove request semantics,
|
||||
instead of parsing provider error strings, choose the transition.
|
||||
|
||||
'''
|
||||
from piker.clearing._state import broker_request
|
||||
|
||||
state: OrderState = OrderState()
|
||||
state.register(order())
|
||||
error: BrokerdError = BrokerdError(
|
||||
account='paper',
|
||||
broker='binance',
|
||||
reason='rejected',
|
||||
request=broker_request(order()),
|
||||
)
|
||||
assert state.accept(error).clients[0].resp == 'error'
|
||||
assert not state.active
|
||||
assert 'one' in state.terminal
|
||||
|
||||
state = OrderState()
|
||||
state.register(order())
|
||||
state.accept(ack())
|
||||
error = error.copy(
|
||||
update={
|
||||
'request': broker_request(order(), 1),
|
||||
}
|
||||
)
|
||||
state.accept(error)
|
||||
assert state.cancel('one').broker[0].reqid == 1
|
||||
Loading…
Reference in New Issue