diff --git a/.claude/skills/verify-tickwright/scripts/strategies.py b/.claude/skills/verify-tickwright/scripts/strategies.py index 7fb10af1..c3eba546 100644 --- a/.claude/skills/verify-tickwright/scripts/strategies.py +++ b/.claude/skills/verify-tickwright/scripts/strategies.py @@ -29,9 +29,9 @@ OrderType, Portfolio, Side, + SignalEmitter, TimeInForce, ) -from tickwright.strategies.emitter import SignalEmitter log = structlog.get_logger() diff --git a/CONTEXT.md b/CONTEXT.md index 073c4c38..626bc865 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -206,7 +206,7 @@ also reads it. Not built yet (#408). _Avoid_: signal uuid, request id. **SignalEmitter**: -The strategy-author's helper (`strategies/emitter.py`) that owns a [[Strategy]]'s monotonic +The strategy-author's helper (`domain/emitter.py`) that owns a [[Strategy]]'s monotonic `seq` counter and builds/publishes its [[Signal]]s (`place`/`cancel`, clock-stamped, returning the [[signal_id]]). Concentrates the one piece of strategy mechanics that is a correctness spine — never reusing a [[signal_id]] across restart — so it is not each author's problem. A strategy diff --git a/docs/adr/0060-what-an-exit-run-reports.md b/docs/adr/0060-what-an-exit-run-reports.md index 3f16b615..38daa2b6 100644 --- a/docs/adr/0060-what-an-exit-run-reports.md +++ b/docs/adr/0060-what-an-exit-run-reports.md @@ -95,6 +95,16 @@ when a venue read fails, and which named events an exit run emits. Decided in its own `left`, because only the job knows what its last read covered. The engine writes only `unread`. Flatten's `left` lists the resting orders first, then each position as `SYMBOL=size` with the signed size, in one comma list. For example `8812,BTC=0.5`.**)** + + **(Field names as shipped, since [#473](https://github.com/MarcosACH/tickwright/issues/473):** + `size` on `flatten.gave_up` is the last size the venue reported for the symbol, signed, as a + string. After failed reads it can be from before the last close, so it can be more than is left. + `exit.finished` says what is left. `count` on + `flatten.dry_attempt` is the dry attempts in a row so far, from 1 to 3.**)** + + **(Extended by [#473](https://github.com/MarcosACH/tickwright/issues/473):** `reason` on + `flatten.dry_attempt` can also be `size_gap`. That is a gap between the partitions and the venue + size that a drain did not close (ADR-0054). No close goes out against it.**)** - **No separate summary.** `exit.finished` is the summary. It goes through the same logger as every other event, so it is the last line the operator sees. diff --git a/docs/extending.md b/docs/extending.md index 9900662b..353aa2ea 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -82,7 +82,7 @@ A strategy consumes ticks and lifecycle events and emits `PlaceSignal`/`CancelSi - [ ] Add a class in `src/tickwright/strategies/` implementing the `Strategy` Protocol: `strategy_id`, `on_tick`, `on_order_event`, `set_next_seq`, `snapshot`, `restore`. - [ ] Emit signals through a composed `SignalEmitter` - ([`strategies/emitter.py`](../src/tickwright/strategies/emitter.py)) — it owns the `seq`, + ([`domain/emitter.py`](../src/tickwright/domain/emitter.py)) — it owns the `seq`, clock-stamps each signal, and publishes it. Never build a `signal_id` by hand. - [ ] Close a position with a reduce-only order. It can only shrink the account net, never grow it or flip it. An order larger than the net is shrunk to it, and the cut part ends `CANCELLED` diff --git a/docs/module-maps/safe-exit.md b/docs/module-maps/safe-exit.md index 42c0b4d6..65c56e3e 100644 --- a/docs/module-maps/safe-exit.md +++ b/docs/module-maps/safe-exit.md @@ -36,7 +36,7 @@ src/tickwright/ store/ # the lock, the seq record, split_basis feed/replay.py # add_symbols is a no-op venues/hyperliquid/ # r flag, batches of 200, refusals, mark cap, reads, feed symbols - strategies/emitter.py # cancel_all(), reduce_only on place() + domain/emitter.py # cancel_all(), reduce_only on place(), moved from strategies observability/catalog.py # the 8 new events app/ __main__.py # cancel-all and flatten commands, --new-store @@ -156,6 +156,13 @@ context. Starts the feed through the `FeedHandle`. - **(Deferred by #472:** the closes run one after another, and there is no pass lock. Each symbol gets one attempt, so only one pass runs at a time anyway. #473 builds the loops and the lock, because retries need them.**)** +- **(Built by #473:** each symbol runs in its own task in one `asyncio.TaskGroup`, so a fault in + one loop cancels the others. The pass lock is an `asyncio.Lock` that every loop shares. A loop + returns nothing. Only the final read says what is left.**)** +- **(Built by #473:** a close takes the same lock as a pass. A publish made while another task's + cascade runs joins that cascade and returns before it is handled. The loop would then read no + saga as a close that filled nothing, and send a second one beside it. A close that still has no + saga after its publish stops the run, so the next boot resumes it.**)** - Do the final venue read. Done means no resting order and no position. - Keep the `__operator__` seq counter. Start it from the same saga high-water fold the `StrategyHost` uses. `engine` cannot import `SignalEmitter`. @@ -164,6 +171,9 @@ context. Starts the feed through the `FeedHandle`. flatten closed. It exits 1 and names the symbol in `left`. #474 builds the counter.**)** - **(Built by #474:** the fold moved to `Cache.seq_high_water()`. `StrategyHost` and `Flatten` both read it. Each close takes the next seq.**)** +- **(Built by #473:** `SignalEmitter` moved to `domain`, so `engine` can import it. `Flatten` keeps + no counter of its own. It seeds an emitter for `__operator__` the way `StrategyHost` seeds a + strategy's, and every close goes through `place()`. The seq rule (ADR-0016) lives in one place.**)** **The per-symbol loop (private):** @@ -174,6 +184,9 @@ context. Starts the feed through the `FeedHandle`. played. Its passes move the replay clock, so an earlier wait would fault the replay. The size is read again after the wait, because the old close may have filled. A wait that ends with the saga still open sends no close for that symbol. The final read reports its size.**)** +- **(Built by #473:** the resume wait and the read after it are the ones an attempt uses. So a + failed pass or a failed read there is a dry attempt, in the same count as the loop's. A flat + read there moves the leftover too.**)** - Place only when the partition sizes sum to the venue size from `fetch_positions()` (ADR-0054). If they differ, call `bus.drain()` first, so every fill already published is applied. If they still differ and the venue has account truth, run `LedgerReconciliation.reconcile_account()`. @@ -185,9 +198,24 @@ context. Starts the feed through the `FeedHandle`. - **(Deferred by #472:** this size check and the drain moved to #475. On paper the sizes always match when flatten places. #472 also sends no order for a symbol the feed never priced, and the final read reports it. Under replay no later price can come.**)** +- **(Built by #473:** the size check and the drain. A gap the drain does not close is a dry + attempt with reason `size_gap`. The gap is checked again after a new venue read. The check + compares the venue size with the account net size from `portfolio.account_net()`.**)** +- **(Built by #473:** every attempt reads the venue before it counts, even one that sent no + close for a missing price or a gap. So a venue that went flat ends the loop with no dry + attempt.**)** +- **(Deferred by #473:** the `reconcile_account()` step moved to #475, code and test together. + Paper has no account truth, so it cannot run there, and a live venue arrives in #475.**)** - A gap that is still there counts as a dry attempt. So a gap that never heals ends in `flatten.gave_up`, not in a loop. - A flat venue moves each leftover into the unattributed partition instead. +- **(Built by #473:** `Checkpointer.checkpoint_leftover()` does the move in one transaction, with + no order. Each strategy leaves at its own entry price. The unattributed side is one piece at the + cost the strategies paid together, so no cash moves. With strategies of opposite sign, that price + is a cost figure and can be zero or negative. Only a read inside a loop triggers it.**)** +- **(Deferred by #473:** a symbol the venue already reads flat at the boot read gets no loop, so + its leftover does not move. [#489](https://github.com/MarcosACH/tickwright/issues/489) builds + it.**)** - Place one reduce-only market IOC order, as a `PlaceSignal` from `__operator__` on the bus. Size it from `fetch_positions()`. - After each attempt, run passes until the saga has a verdict. Wait the in-flight interval between @@ -299,7 +327,7 @@ check. `__operator__` owns no partition, so that check would deny every flatten --- -### StrategyHost and SignalEmitter (`engine`, `strategies`, existing) +### StrategyHost and SignalEmitter (`engine`, `domain`, existing) **Interface:** The emitter gains `cancel_all(symbol=...)`, which returns its `signal_id`. `place()` gains `reduce_only`. @@ -311,6 +339,9 @@ a handled cancel all's seq (ADR-0055). **Depth note:** The seq stays owned by the emitter. The host only resumes it. +**(Built by #473:** the emitter moved from `strategies` to `domain`. `Flatten` uses one for +`__operator__`.**)** + --- ### Exchange (`domain`, existing seam) diff --git a/docs/module-maps/v1-core-engine.md b/docs/module-maps/v1-core-engine.md index 5a8d1d08..2b732ce5 100644 --- a/docs/module-maps/v1-core-engine.md +++ b/docs/module-maps/v1-core-engine.md @@ -195,7 +195,7 @@ src/tickwright/ ### strategies (`strategies/`) -**Interface:** Minimal reference `Strategy` adapters (engine capability, not a library). Authors must know: `on_tick`/`on_order_event`; emit `PlaceSignal`/`CancelSignal` via a composed `SignalEmitter` (`emitter.py`) that owns the strategy-owned monotonic `seq` (resumed at startup via the engine-set `set_next_seq()`, ADR-0016) and clock-stamps + publishes each signal; own state *content* via `snapshot()`/`restore()` (engine persists the bytes); cancel by your own `signal_id`; handler exceptions are contained, not fatal. +**Interface:** Minimal reference `Strategy` adapters (engine capability, not a library). Authors must know: `on_tick`/`on_order_event`; emit `PlaceSignal`/`CancelSignal` via a composed `SignalEmitter` (`domain/emitter.py`) that owns the strategy-owned monotonic `seq` (resumed at startup via the engine-set `set_next_seq()`, ADR-0016) and clock-stamps + publishes each signal; own state *content* via `snapshot()`/`restore()` (engine persists the bytes); cancel by your own `signal_id`; handler exceptions are contained, not fatal. **Responsibilities:** Signal logic only. No persistence, no venue knowledge, no saga awareness beyond `OrderEvent`s. diff --git a/src/tickwright/domain/__init__.py b/src/tickwright/domain/__init__.py index 2269428c..2c6f1f61 100644 --- a/src/tickwright/domain/__init__.py +++ b/src/tickwright/domain/__init__.py @@ -8,6 +8,7 @@ from .account import Account, AccountSpec, AccountView, venue_cash from .duration import duration_ns from .economics import fill_fee, funding_amount, funding_boundaries +from .emitter import SignalEmitter from .enums import ( AggressorSide, ComponentState, @@ -231,6 +232,8 @@ "account_valuation", "account_view", "position_view", + # the one seq counter for every signal source (ADR-0016) + "SignalEmitter", # retry pacing "Backoff", "Deadline", diff --git a/src/tickwright/strategies/emitter.py b/src/tickwright/domain/emitter.py similarity index 92% rename from src/tickwright/strategies/emitter.py rename to src/tickwright/domain/emitter.py index ed4ccebe..54a601a2 100644 --- a/src/tickwright/strategies/emitter.py +++ b/src/tickwright/domain/emitter.py @@ -7,20 +7,15 @@ it, plus the clock-stamped envelope and the publish, behind two typed methods. A strategy *composes* an emitter (holds one as a field) rather than inheriting a base: the ``Strategy`` seam is satisfied by shape, never by inheritance (ADR-0032). -Depends on ``domain`` only, so it emits no named events. +It lives in ``domain`` so the engine's own operator jobs keep their seq the same +way. One counter rule then serves every signal source. It emits no named events. """ from decimal import Decimal -from tickwright.domain import ( - CancelSignal, - Clock, - EventBus, - OrderType, - PlaceSignal, - Side, - TimeInForce, -) +from .enums import OrderType, Side, TimeInForce +from .events import CancelSignal, PlaceSignal +from .protocols import Clock, EventBus class SignalEmitter: diff --git a/src/tickwright/engine/checkpoint.py b/src/tickwright/engine/checkpoint.py index ba68e685..e6ebd07e 100644 --- a/src/tickwright/engine/checkpoint.py +++ b/src/tickwright/engine/checkpoint.py @@ -311,6 +311,46 @@ def _split( for owner, move in moves.items() ] + def checkpoint_leftover(self, symbol: str) -> None: + """Move every strategy's size on ``symbol`` into the unattributed partition. + + Only for a venue that reads flat, where no order can close the size + (ADR-0054). Each strategy leaves at its own entry price, so its realized + PnL does not change. The unattributed side is one piece at the cost the + strategies paid together, so no cash moves (ADR-0054). With strategies of + opposite sign, that price is a cost figure, not a traded price. It can be + zero or negative. No order and no fee, and every partition moves in one + transaction. + """ + ts_ns = self._clock.timestamp_ns() + pieces: list[tuple[PartitionFill, Side]] = [] + moved = cost = Decimal(0) + for owner, size in self._portfolio.partition_sizes(symbol): + view = self._portfolio.position(symbol, strategy_id=owner) + if owner is None or not size or view is None: + continue + pieces.append( + _leftover_piece(owner, symbol, -size, price=view.entry_price, ts_ns=ts_ns) + ) + moved += size + cost += size * view.entry_price + if moved: + pieces.append(_leftover_piece(None, symbol, moved, price=cost / moved, ts_ns=ts_ns)) + if not pieces: + return + changes = [self._portfolio.apply_fill(piece, side=side) for piece, side in pieces] + try: + self._store.checkpoint_ledger( + account=changes[0].account, + positions=tuple(change.position for change in changes), + ts_ns=ts_ns, + ) + except InvariantViolation as exc: + # Only the seam's own failure type is relabelled, as its siblings do. + raise InvariantViolation(f"leftover move write failed for {symbol}") from exc + for change in changes: + self._portfolio.project(change) + def checkpoint_heal( self, fills: Sequence[ReconciliationFill], @@ -459,3 +499,20 @@ def checkpoint_funding(self, accrual: FundingAccrual) -> None: f"at boundary {accrual.boundary_ts_ns}" ) from exc self._portfolio.project_funding(change) + + +def _leftover_piece( + owner: str | None, symbol: str, size: Decimal, *, price: Decimal, ts_ns: int +) -> tuple[PartitionFill, Side]: + """One partition's signed ``size`` move of a leftover, at ``price`` and with no fee.""" + return ( + PartitionFill( + event_id=f"leftover:{symbol}:{ts_ns}:{UNATTRIBUTED if owner is None else owner}", + strategy_id=owner, + symbol=symbol, + quantity=abs(size), + price=price, + fee=Decimal(0), + ), + Side.BUY if size > 0 else Side.SELL, + ) diff --git a/src/tickwright/engine/exit_run.py b/src/tickwright/engine/exit_run.py index 73faa6fe..2ac2e764 100644 --- a/src/tickwright/engine/exit_run.py +++ b/src/tickwright/engine/exit_run.py @@ -6,6 +6,7 @@ sees them. """ +import asyncio from collections.abc import Sequence from dataclasses import dataclass from decimal import Decimal @@ -22,9 +23,9 @@ OrderAnchor, OrderRef, OrderType, - PlaceSignal, PreTradeGuard, Side, + SignalEmitter, TimeInForce, VenueOpenOrder, VenueReadFailure, @@ -138,90 +139,278 @@ async def note_price(tick: MarketTick) -> None: # Only a replay ends, and it plays the whole file first. So each close # fills at the last row (ADR-0059). The CLI refuses a live feed until #475. await context.feed.ended() - # From the sagas, as a strategy's is. An earlier flatten's seq is - # consumed, and the manager drops a signal id it has seen (ADR-0006). - seq = context.checkpointer.cache.seq_high_water().get(OPERATOR, 0) - for symbol, size in positions.items(): - # No venue can send a market close without a price. The symbol - # stays open, and the final read reports its size. - if symbol not in priced: - continue - resumed = await _resume_closes(symbol, size, context) - if isinstance(resumed, VenueReadFailure): - return ExitOutcome(status=ExitStatus.STOPPED, left=describe_failure(resumed)) - if resumed: - seq += 1 - await _await_verdict(await _close(symbol, resumed, seq, context), context) + emitter = SignalEmitter(strategy_id=OPERATOR, bus=context.bus, clock=context.clock) + # Seeded from the sagas, as a strategy's is. An earlier flatten's seq + # is consumed, and the manager drops a signal id it has seen (ADR-0006). + high_water = context.checkpointer.cache.seq_high_water() + emitter.set_next_seq(high_water.get(OPERATOR, 0) + 1) + turns = _Turns(context.reconciler) + # One loop per symbol, all at the same time, so a thin book on one + # symbol does not hold back the others (ADR-0056). A task group + # cancels the rest when one faults. A loop reports nothing, because + # only the final read can say what is left (ADR-0060). + async with asyncio.TaskGroup() as group: + for symbol, size in positions.items(): + group.create_task( + _flatten_symbol(symbol, size, priced, emitter, turns, context) + ) return await _final_read(context) +DRY_ATTEMPTS_TO_GIVE_UP = 3 +"""Dry attempts in a row that stop a symbol's loop (ADR-0056).""" + +DRY_WAIT_SECONDS = 2.0 +"""The wait after a dry attempt, before the next one (ADR-0056).""" + + +class _Turns: + """One close or reconcile pass at a time, shared by every loop of one flatten run. + + A loop that needs a turn waits for the one in progress, then takes its own. + A pass that started before the loop's attempt ended may not cover it. + + A close takes a turn too. A publish made while another task's cascade runs + joins that cascade and returns before it is handled. The loop would then + find no saga, read it as a close that filled nothing, and send a second + one beside it (ADR-0056). + """ + + def __init__(self, reconciler: Reconciler) -> None: + self._reconciler = reconciler + self._lock = asyncio.Lock() + + async def run_pass(self) -> bool: + """Run both halves of one pass. False when either failed.""" + async with self._lock: + return ( + await self._reconciler.reconcile_inflight() + and await self._reconciler.reconcile_open_orders() + ) + + async def close(self, symbol: str, size: Decimal, emitter: SignalEmitter) -> str: + """Send one close, and return once the engine has handled it.""" + async with self._lock: + return await _close(symbol, size, emitter) + + +class _DryAttempts: + """One symbol's count of attempts in a row that closed nothing (ADR-0056).""" + + def __init__(self, symbol: str, clock: Clock) -> None: + self._symbol = symbol + self._clock = clock + self._count = 0 + + def reset(self) -> None: + self._count = 0 + + async def gave_up(self, reason: str, size: Decimal) -> bool: + """Count one dry attempt. Say whether it was the last, or wait before the next. + + ``size`` is the last size the venue reported. After failed reads it can be + from before the last close, so the final read says what is left (ADR-0060). + """ + self._count += 1 + named_event( + NamedEvent.FLATTEN_DRY_ATTEMPT, symbol=self._symbol, count=self._count, reason=reason + ) + if self._count == DRY_ATTEMPTS_TO_GIVE_UP: + named_event(NamedEvent.FLATTEN_GAVE_UP, symbol=self._symbol, size=str(size)) + return True + await self._clock.sleep(DRY_WAIT_SECONDS) + return False + + +async def _flatten_symbol( + symbol: str, + size: Decimal, + priced: set[str], + emitter: SignalEmitter, + turns: _Turns, + context: ExitContext, +) -> None: + """One symbol's loop: wait for a close the boot left open, then close the rest.""" + # One count for the whole loop, so the resume wait's dry attempts are in the row. + dry = _DryAttempts(symbol, context.clock) + resumed = await _resume_closes(symbol, size, dry, turns, context) + if resumed is not None: + await _close_symbol(symbol, resumed, priced, emitter, dry, turns, context) + + +async def _close_symbol( + symbol: str, + size: Decimal, + priced: set[str], + emitter: SignalEmitter, + dry: _DryAttempts, + turns: _Turns, + context: ExitContext, +) -> None: + """Close ``symbol`` until the venue reads it flat, or stop after 3 dry attempts in a row. + + A fill resets the count, so a thin book that keeps filling a little never gives up + (ADR-0056). A failed read is dry too, so a venue that cannot answer never hangs + the run (ADR-0060). The final read reports what is left. + """ + while attempt := await _attempt(symbol, size, priced, emitter, dry, turns, context): + size = attempt.left + if attempt.dry is None: + dry.reset() + elif await dry.gave_up(attempt.dry, size): + return + + +@dataclass(frozen=True, slots=True, kw_only=True) +class _Attempt: + """How one attempt ended, when the loop goes on after it.""" + + left: Decimal + """What the venue still holds on the symbol. Never zero.""" + dry: str | None + """Why the attempt closed nothing, or ``None`` when it filled.""" + + +async def _attempt( + symbol: str, + size: Decimal, + priced: set[str], + emitter: SignalEmitter, + dry: _DryAttempts, + turns: _Turns, + context: ExitContext, +) -> _Attempt | None: + """Try to close ``size`` once, then read what is left. ``None`` ends the loop. + + Every attempt reads the venue before it counts. So a venue that went flat + ends the loop with no dry attempt, even when no close went out (ADR-0056). + """ + reason: str | None + # No venue can send a market close without a price. + if symbol not in priced: + reason = "no_price" + elif await _size_gap(symbol, size, context): + reason = "size_gap" + else: + cloid = await turns.close(symbol, size, emitter) + if not await _verdict(cloid, size, dry, turns, context): + return None + reason = None if _filled(cloid, context) else "no_fill" + left = await _read_left(symbol, size, dry, context) + return None if left is None else _Attempt(left=left, dry=reason) + + +async def _size_gap(symbol: str, size: Decimal, context: ExitContext) -> bool: + """Whether the account net size on ``symbol`` differs from the venue's ``size``. + + A close is split across the partitions it closes (ADR-0054). Against a gap, + some of it would land on no partition, or one would be left holding size. + The gap can be fills the bus has not delivered yet, so a drain comes first. + """ + if _held(symbol, context) == size: + return False + await context.bus.drain() + return _held(symbol, context) != size + + +def _held(symbol: str, context: ExitContext) -> Decimal: + """The account net size on ``symbol``, from the portfolio's one definition of it.""" + return context.checkpointer.portfolio.account_net().get(symbol, Decimal(0)) + + +async def _verdict( + cloid: str, size: Decimal, dry: _DryAttempts, turns: _Turns, context: ExitContext +) -> bool: + """Run reconcile passes until the close's saga is terminal (ADR-0011, ADR-0026). + + A pass covers both halves. ``reconcile_inflight`` resolves a send that never + acked. ``reconcile_open_orders`` resolves an acked order whose last push was lost. + + A failed pass is a dry attempt, so a venue that cannot answer never hangs the + run (ADR-0060). False when failed passes used up the dry attempts. A second + close beside an open one could close the size twice. So a failed pass waits + for this close again, and a give-up leaves it to the next boot. + """ + while not _decided(cloid, context): + await context.clock.sleep(context.pass_interval_seconds) + if not await turns.run_pass() and await dry.gave_up("read_failed", size): + return False + return True + + +async def _read_left( + symbol: str, held: Decimal, dry: _DryAttempts, context: ExitContext +) -> Decimal | None: + """What the venue still holds on ``symbol``, read again after each failed read. + + ``None`` ends the loop. Either the failed reads used up the dry attempts, or + the venue reads flat. ``held`` is the last size read, for ``flatten.gave_up``. + + A hand trade the books cannot see may have closed a flat venue, and no order + can close what the venue does not hold. So what the strategies still hold + moves to the unattributed partition (ADR-0054, ADR-0056). + """ + while isinstance(positions := await context.account.fetch_positions(), VenueReadFailure): + if await dry.gave_up("read_failed", held): + return None + left = positions.get(symbol, Decimal(0)) + if not left: + context.checkpointer.checkpoint_leftover(symbol) + return None + return left + + async def _resume_closes( - symbol: str, size: Decimal, context: ExitContext -) -> Decimal | VenueReadFailure: + symbol: str, size: Decimal, dry: _DryAttempts, turns: _Turns, context: ExitContext +) -> Decimal | None: """Wait for any close the boot left open in ``symbol``, then say what is left to close. That close could still fill, so a second one sent beside it could close the - size twice. A symbol never has two closes open (ADR-0056). So a wait that ends - with one still open leaves nothing to close. The final read reports the size, - and the next boot resumes the saga. The wait runs after the feed has played, - because its passes move the replay clock. + size twice. A symbol never has two closes open (ADR-0056). So the wait and the + read are the same ones that follow an attempt. ``None`` ends the loop, as it + does there. After a give-up, the final read reports the size, and the next boot + resumes the saga. The wait runs after the feed has played, because its passes + move the replay clock. """ left_open = context.checkpointer.cache.open_orders(strategy_id=OPERATOR, symbol=symbol) if not left_open: return size for order in left_open: - await _await_verdict(order.cloid, context) - if not all(_decided(order.cloid, context) for order in left_open): - return Decimal(0) - positions = await context.account.fetch_positions() - if isinstance(positions, VenueReadFailure): - return positions - return positions.get(symbol, Decimal(0)) + if not await _verdict(order.cloid, size, dry, turns, context): + return None + # The old close may have filled during the wait, so only a new read sizes the next. + return await _read_left(symbol, size, dry, context) -async def _close(symbol: str, size: Decimal, seq: int, context: ExitContext) -> str: +async def _close(symbol: str, size: Decimal, emitter: SignalEmitter) -> str: """Send one reduce-only market IOC order for the whole of ``size`` (ADR-0054). Returns the cloid of the order's saga. """ - now = context.clock.timestamp_ns() - signal = PlaceSignal( - ts_event=now, - ts_init=now, - strategy_id=OPERATOR, + signal_id = await emitter.place( symbol=symbol, - seq=seq, side=Side.SELL if size > 0 else Side.BUY, quantity=abs(size), order_type=OrderType.MARKET, time_in_force=TimeInForce.IOC, reduce_only=True, ) - await context.bus.publish(signal) - return derive_cloid(signal.signal_id) - + return derive_cloid(signal_id) -async def _await_verdict(cloid: str, context: ExitContext) -> None: - """Run reconcile passes until the close's saga is terminal (ADR-0011, ADR-0026). - - A pass covers both halves. ``reconcile_inflight`` resolves a send that never - acked. ``reconcile_open_orders`` resolves an acked order whose last push was lost. - A failed pass ends the wait, so a venue that cannot answer never hangs the run. - The saga stays open, and the next boot resumes it. - """ - while not _decided(cloid, context): - await context.clock.sleep(context.pass_interval_seconds) - if not await context.reconciler.reconcile_inflight(): - return - if not await context.reconciler.reconcile_open_orders(): - return +def _decided(cloid: str, context: ExitContext) -> bool: + saga = context.checkpointer.cache.get_order(cloid) + if saga is None: + # The manager writes a saga for every signal it handles, a denial included. + # So the close was never handled, and it may still reach the venue. A + # second close beside it could close the size twice (ADR-0056). + raise InvariantViolation(f"flatten close {cloid} was never handled") + return saga.is_terminal -def _decided(cloid: str, context: ExitContext) -> bool: +def _filled(cloid: str, context: ExitContext) -> bool: saga = context.checkpointer.cache.get_order(cloid) - # The manager writes a saga for every signal it handles, a denial included. - return saga is None or saga.is_terminal + return saga is not None and saga.cum_qty > 0 async def _final_read(context: ExitContext) -> ExitOutcome: diff --git a/src/tickwright/observability/catalog.py b/src/tickwright/observability/catalog.py index 64edd6b5..91cf315d 100644 --- a/src/tickwright/observability/catalog.py +++ b/src/tickwright/observability/catalog.py @@ -65,6 +65,8 @@ class NamedEvent(StrEnum): EXIT_FINISHED = "exit.finished" EXIT_UNGUARDED = "exit.unguarded" CANCEL_ALL_ORDERS_REMAIN = "cancel_all.orders_remain" + FLATTEN_DRY_ATTEMPT = "flatten.dry_attempt" + FLATTEN_GAVE_UP = "flatten.gave_up" # Pre-trade kill switch (``PreTradeGuard``). GUARD_KILL_SWITCH_TRIPPED = "guard.kill_switch_tripped" @@ -280,6 +282,8 @@ class NamedEvent(StrEnum): NamedEvent.EXIT_FINISHED: frozenset({"job", "outcome", "exit_code", "left"}), NamedEvent.EXIT_UNGUARDED: frozenset({"warning"}), NamedEvent.CANCEL_ALL_ORDERS_REMAIN: frozenset({"orders"}), + NamedEvent.FLATTEN_DRY_ATTEMPT: frozenset({"symbol", "count", "reason"}), + NamedEvent.FLATTEN_GAVE_UP: frozenset({"symbol", "size"}), NamedEvent.GUARD_KILL_SWITCH_TRIPPED: frozenset({"reason"}), NamedEvent.GUARD_KILL_SWITCH_RESET: frozenset(), NamedEvent.STRATEGY_ERROR: frozenset({"strategy_id", "event_id", "error"}), diff --git a/src/tickwright/strategies/__init__.py b/src/tickwright/strategies/__init__.py index 244381f5..4dc6379b 100644 --- a/src/tickwright/strategies/__init__.py +++ b/src/tickwright/strategies/__init__.py @@ -2,8 +2,7 @@ capability, not a library; each shipped impl only proves the seam. This package stays domain-only — a strategy emits no named events.""" -from .emitter import SignalEmitter from .single_shot import SingleShotMarketStrategy from .single_shot_limit import SingleShotLimitStrategy -__all__ = ["SignalEmitter", "SingleShotLimitStrategy", "SingleShotMarketStrategy"] +__all__ = ["SingleShotLimitStrategy", "SingleShotMarketStrategy"] diff --git a/src/tickwright/strategies/single_shot.py b/src/tickwright/strategies/single_shot.py index 8c2e6e8b..ee3512b9 100644 --- a/src/tickwright/strategies/single_shot.py +++ b/src/tickwright/strategies/single_shot.py @@ -26,11 +26,10 @@ Portfolio, PositionView, Side, + SignalEmitter, TimeInForce, ) -from .emitter import SignalEmitter - class SingleShotMarketStrategy: """Buys (or sells) a fixed quantity at market, once, on the first tick seen.""" diff --git a/src/tickwright/strategies/single_shot_limit.py b/src/tickwright/strategies/single_shot_limit.py index c34ef554..d904baf0 100644 --- a/src/tickwright/strategies/single_shot_limit.py +++ b/src/tickwright/strategies/single_shot_limit.py @@ -23,11 +23,10 @@ OrderFilled, OrderType, Side, + SignalEmitter, TimeInForce, ) -from .emitter import SignalEmitter - class SingleShotLimitStrategy: """Places one resting LIMIT on the first tick; optionally cancels it later.""" diff --git a/tests/app/test_pre_trade_limits_e2e.py b/tests/app/test_pre_trade_limits_e2e.py index 73da1b3d..2fab08fe 100644 --- a/tests/app/test_pre_trade_limits_e2e.py +++ b/tests/app/test_pre_trade_limits_e2e.py @@ -52,11 +52,11 @@ OrderType, Portfolio, Side, + SignalEmitter, TimeInForce, ) from tickwright.engine.guard import PreTradeLimits, RateCap, SymbolLimits from tickwright.engine.runner import Engine -from tickwright.strategies.emitter import SignalEmitter SECOND_NS = 1_000_000_000 PRICES = {"BTC": Decimal("50000"), "ETH": Decimal("3000")} diff --git a/tests/app/test_reduce_only_e2e.py b/tests/app/test_reduce_only_e2e.py index fa20305a..d03a05ab 100644 --- a/tests/app/test_reduce_only_e2e.py +++ b/tests/app/test_reduce_only_e2e.py @@ -43,10 +43,10 @@ OrderType, Portfolio, Side, + SignalEmitter, TimeInForce, ) from tickwright.engine.runner import Engine -from tickwright.strategies.emitter import SignalEmitter SECOND_NS = 1_000_000_000 PRICE = Decimal("50000") diff --git a/tests/strategies/test_emitter.py b/tests/domain/test_emitter.py similarity index 98% rename from tests/strategies/test_emitter.py rename to tests/domain/test_emitter.py index 978e63cb..5e71edb3 100644 --- a/tests/strategies/test_emitter.py +++ b/tests/domain/test_emitter.py @@ -18,9 +18,9 @@ PlaceSignal, Side, Signal, + SignalEmitter, TimeInForce, ) -from tickwright.strategies import SignalEmitter def _harness() -> tuple[InMemoryBus, ManualClock, SignalEmitter, list[Signal]]: diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 4c1a9f44..a676a730 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -8,6 +8,7 @@ import asyncio import json from collections.abc import Callable, Iterable, Sequence +from dataclasses import replace from decimal import Decimal from functools import partial from pathlib import Path @@ -27,13 +28,17 @@ OPERATOR, AggressorSide, Clock, + Event, + EventBus, Exchange, ExecutionReport, + Handler, InstrumentSpec, MarketTick, Order, OrderRef, OrderState, + OrderStatusReport, OrderType, PlaceOrder, PlaceSignal, @@ -79,7 +84,7 @@ def _ticks(path: Path, replayed: Iterable[str]) -> Path: return path -def _paper(bus: InMemoryBus, clock: ManualClock, store: SQLiteStore) -> PaperExchange: +def _paper(bus: EventBus, clock: ManualClock, store: SQLiteStore) -> PaperExchange: reads = paper_store_reads(store) return PaperExchange( bus=bus, @@ -92,8 +97,8 @@ def _paper(bus: InMemoryBus, clock: ManualClock, store: SQLiteStore) -> PaperExc ) -def _hold(path: Path, sizes: dict[str, str], *, strategy_id: str = "trivial") -> None: - """Open each signed size as ``strategy_id``, at the symbol's first row price. +def _hold(path: Path, sizes: dict[str, str], *, strategy_id: str = "trivial", row: int = 0) -> None: + """Open each signed size as ``strategy_id``, at the price of the symbol's ``row``. Seeded through the engine's own order path on a separate bus, as an earlier run would. """ @@ -109,7 +114,7 @@ def _hold(path: Path, sizes: dict[str, str], *, strategy_id: str = "trivial") -> async def go() -> None: for seq, (symbol, size) in enumerate(sizes.items(), start=1): - await bus.publish(_tick(symbol, _ROWS[symbol][0])) + await bus.publish(_tick(symbol, _ROWS[symbol][row])) await bus.publish( PlaceSignal( ts_event=0, @@ -147,16 +152,20 @@ def _flatten( link: Callable[[PaperExchange], Exchange] | None = None, replayed: Iterable[str] = tuple(_ROWS), resting: bool = False, + clock: ManualClock | None = None, + bus: EventBus | None = None, ) -> tuple[int, list[EventDict]]: """Run flatten over paper and the ``_ROWS`` replay file. ``guarded`` gives the engine a real guard. ``link`` puts a venue double in front of paper. ``replayed`` picks the symbols the file has rows for. ``resting`` rests one BTC buy far below the replay, as ``_RESTING``. + ``clock`` lets a test read the virtual time the run ended at. ``bus`` lets + a venue double publish its own reports. Returns the exit code and every event the run emitted. """ - bus = InMemoryBus() - clock = ManualClock() + bus = bus or InMemoryBus() + clock = clock or ManualClock() store = SQLiteStore(tmp_path / "saga.db") paper = _paper(bus, clock, store) engine = Engine( @@ -182,7 +191,7 @@ async def go() -> int: _RESTING = derive_cloid("resting:BTC:1") -async def _rest(bus: InMemoryBus, paper: PaperExchange, store: SQLiteStore, clock: Clock) -> None: +async def _rest(bus: EventBus, paper: PaperExchange, store: SQLiteStore, clock: Clock) -> None: """Rest ``_RESTING`` on paper, with its saga in the store. Paper keeps its book in memory only (#484), so the order goes on the run's own @@ -279,21 +288,291 @@ def test_flatten_closes_a_long_at_the_last_replay_row_and_the_partition_reads_fl assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} -def test_a_held_symbol_the_replay_never_prices_stops_the_run_with_its_size_left( +def test_a_held_symbol_the_replay_never_prices_gives_up_after_three_dry_attempts( tmp_path: Path, ) -> None: - # Paper has no price to fill BTC at, so the close cannot fill. The final read says so. + # No price can come after the replay ends, so every attempt is dry (ADR-0056). + # The replay ends at 3000 ns. Two 2 second waits sit between the three attempts. _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + clock = ManualClock() - exit_code, logs = _flatten(tmp_path, replayed=("ETH",)) + exit_code, logs = _flatten(tmp_path, replayed=("ETH",), clock=clock) assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["symbol"], log["count"], log["reason"]) for log in dry] == [ + ("BTC", 1, "no_price"), + ("BTC", 2, "no_price"), + ("BTC", 3, "no_price"), + ] + gave_up = [log for log in logs if log["event"] == "flatten.gave_up"] + assert [(log["symbol"], log["size"]) for log in gave_up] == [("BTC", "0.02")] + assert clock.timestamp_ns() == 3_000 + 4_000_000_000 assert logs[-1]["event"] == "exit.finished" assert logs[-1]["outcome"] == "stopped" assert logs[-1]["left"] == "BTC=0.02" assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0.02"), Decimal("0"))} +class _ThinBook(VenueLink): + """Each close fills only what the book holds, and the venue cancels the rest. + + ``fills`` gives the size each send finds, in order. Zero fills nothing. + """ + + def __init__(self, venue: Exchange, *, bus: EventBus, fills: Sequence[str]) -> None: + super().__init__(venue) + self._bus = bus + self._fills = [Decimal(fill) for fill in fills] + + async def place(self, order: PlaceOrder) -> None: + fill = self._fills.pop(0) + if fill: + await super().place(replace(order, quantity=fill)) + # An IOC never rests, so whatever did not fill ends cancelled. + await self._bus.publish( + OrderStatusReport( + ts_event=0, + ts_init=0, + cloid=order.cloid, + symbol=order.symbol, + status=OrderState.CANCELLED, + ) + ) + + +def test_a_fill_between_dry_attempts_resets_the_count(tmp_path: Path) -> None: + # Two dry attempts, then a partial fill, then three more dry ones. Without the + # reset, flatten would give up on the third dry attempt, before its fifth send. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + bus = InMemoryBus() + link = partial(_ThinBook, bus=bus, fills=("0", "0", "0.01", "0", "0", "0")) + + exit_code, logs = _flatten(tmp_path, link=link, bus=bus) + + assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [ + (1, "no_fill"), + (2, "no_fill"), + (1, "no_fill"), + (2, "no_fill"), + (3, "no_fill"), + ] + gave_up = [log for log in logs if log["event"] == "flatten.gave_up"] + assert [(log["symbol"], Decimal(log["size"])) for log in gave_up] == [("BTC", Decimal("0.01"))] + symbol, left = logs[-1]["left"].split("=") + assert (symbol, Decimal(left)) == ("BTC", Decimal("0.01")) + + +class _ReadsMore(VenueLink): + """The venue reads more size than the ledger holds, by ``extra``, on every read.""" + + def __init__(self, venue: Exchange, *, symbol: str, extra: str) -> None: + super().__init__(venue) + self._symbol = symbol + self._extra = Decimal(extra) + + async def fetch_positions(self) -> dict[str, Decimal] | VenueReadFailure: + positions = await super().fetch_positions() + if isinstance(positions, VenueReadFailure): + return positions + return {**positions, self._symbol: positions.get(self._symbol, Decimal(0)) + self._extra} + + +def test_a_size_gap_that_never_heals_gives_up_with_no_close(tmp_path: Path) -> None: + # A close is split across the partitions, so it may only go out when they sum + # to the venue size (ADR-0054). Paper has no account truth to heal the gap with. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + + exit_code, logs = _flatten(tmp_path, link=partial(_ReadsMore, symbol="BTC", extra="0.01")) + + assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [ + (1, "size_gap"), + (2, "size_gap"), + (3, "size_gap"), + ] + gave_up = [log for log in logs if log["event"] == "flatten.gave_up"] + assert [(log["symbol"], log["size"]) for log in gave_up] == [("BTC", "0.03")] + assert not [log for log in logs if str(log.get("signal_id", "")).startswith(OPERATOR)] + assert logs[-1]["left"] == "BTC=0.03" + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0.02"), Decimal("0"))} + + +class _HoldsOneOrder: + """An ``InMemoryBus`` that holds every report for ``cloid`` until ``drain()``. + + Those are reports the venue published that the engine has not applied yet. + Kafka can leave them in flight, in order, and ``drain()`` waits for them. + """ + + def __init__(self, cloid: str) -> None: + self._bus = InMemoryBus() + self._cloid = cloid + self._held: list[ExecutionReport] = [] + + def subscribe[E: Event](self, event_type: type[E], handler: Handler[E]) -> None: + self._bus.subscribe(event_type, handler) + + async def start(self) -> None: + await self._bus.start() + + async def publish(self, event: Event) -> None: + if isinstance(event, ExecutionReport) and event.cloid == self._cloid: + self._held.append(event) + return + await self._bus.publish(event) + + async def drain(self) -> None: + held, self._held = self._held, [] + for event in held: + await self._bus.publish(event) + await self._bus.drain() + + async def close(self) -> None: + await self._bus.close() + + +class _FillsAsItCancels(VenueLink): + """The market trades through each resting BTC buy just before its cancel lands.""" + + def __init__(self, venue: Exchange, *, bus: EventBus) -> None: + super().__init__(venue) + self._bus = bus + + async def cancel(self, refs: Sequence[OrderRef]) -> None: + await self._bus.publish(_tick("BTC", "30000")) + await super().cancel(refs) + + +def test_a_size_gap_from_a_fill_still_in_flight_closes_after_a_drain(tmp_path: Path) -> None: + # The resting buy fills on the venue as cancel all runs, but its fill is still + # in flight. The drain applies it, so the partitions sum to the venue size + # again, and one close covers both (ADR-0054). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + bus = _HoldsOneOrder(_RESTING) + + exit_code, logs = _flatten( + tmp_path, link=partial(_FillsAsItCancels, bus=bus), bus=bus, resting=True + ) + + assert exit_code == 0 + assert not [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert logs[-1]["outcome"] == "done" + # Both close at 44000. One opened at 42000, the other filled at 30000. + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("resting", "BTC"): (Decimal("0"), Decimal("140")), + } + + +class _SoldByHand(_ThinBook): + """The first close fills nothing, and then the venue reads BTC flat. + + The operator sold it by hand in the venue UI, so the books never see that trade. + """ + + def __init__(self, venue: Exchange, *, bus: EventBus) -> None: + super().__init__(venue, bus=bus, fills=("0",)) + self._sold = False + + async def place(self, order: PlaceOrder) -> None: + await super().place(order) + self._sold = True + + async def fetch_positions(self) -> dict[str, Decimal] | VenueReadFailure: + positions = await super().fetch_positions() + if isinstance(positions, VenueReadFailure) or not self._sold: + return positions + return {symbol: size for symbol, size in positions.items() if symbol != "BTC"} + + +def test_a_venue_that_reads_flat_moves_the_leftover_to_unattributed_at_the_entry_price( + tmp_path: Path, +) -> None: + # No order can close what the venue no longer holds. The strategy's size moves at + # its own entry price, so its realized PnL stays 0. At the 44000 mark it would be + # 40. The unattributed size keeps the gap visible (ADR-0054, ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + bus = InMemoryBus() + + exit_code, logs = _flatten(tmp_path, link=partial(_SoldByHand, bus=bus), bus=bus) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + assert logs[-1]["left"] == "none" + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("0")), + (None, "BTC"): (Decimal("0.02"), Decimal("0")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + unattributed = [row for row in store.all_positions() if row.strategy_id is None] + assert [row.entry_price for row in unattributed] == [Decimal("42000")] + + +class _SoldAfterTheBootRead(VenueLink): + """The venue reads BTC flat on every positions read after flatten's first. + + The operator sold it by hand while flatten waited, so the books never see that trade. + """ + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self._reads = 0 + + async def fetch_positions(self) -> dict[str, Decimal] | VenueReadFailure: + self._reads += 1 + positions = await super().fetch_positions() + if isinstance(positions, VenueReadFailure) or self._reads == 1: + return positions + return {symbol: size for symbol, size in positions.items() if symbol != "BTC"} + + +def test_a_venue_that_reads_flat_with_no_price_ends_the_loop_with_no_give_up( + tmp_path: Path, +) -> None: + # An attempt with no price sends no order, but it still reads the venue before + # it counts. A flat venue ends flatten (ADR-0056), so the leftover moves and + # no dry attempt is counted. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + + exit_code, logs = _flatten(tmp_path, link=_SoldAfterTheBootRead, replayed=("ETH",)) + + assert exit_code == 0 + assert not [log for log in logs if log["event"].startswith("flatten.")] + assert logs[-1]["outcome"] == "done" + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("0")), + (None, "BTC"): (Decimal("0.02"), Decimal("0")), + } + + +def test_two_strategies_of_opposite_sign_leave_one_unattributed_piece_at_their_cost( + tmp_path: Path, +) -> None: + # Each strategy leaves at its own entry, so neither books realized PnL. The + # unattributed side takes their net at the cost they paid together, so no + # cash moves (ADR-0054). That price need not be one the market traded: + # 0.02 at 42000 less 0.01 at 44000 is 400, over a size of 0.01. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + _hold(tmp_path / "saga.db", {"BTC": "-0.01"}, strategy_id="other", row=2) + + exit_code, logs = _flatten(tmp_path, link=_SoldAfterTheBootRead, replayed=("ETH",)) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("0")), + ("other", "BTC"): (Decimal("0"), Decimal("0")), + (None, "BTC"): (Decimal("0.01"), Decimal("0")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + unattributed = [row for row in store.all_positions() if row.strategy_id is None] + assert [row.entry_price for row in unattributed] == [Decimal("40000")] + + def test_flatten_cancels_a_resting_order_before_it_closes_a_long_and_a_short( tmp_path: Path, ) -> None: @@ -375,7 +654,8 @@ async def fetch_positions(self) -> dict[str, Decimal] | VenueReadFailure: [ pytest.param("fetch_positions", 1, id="positions-before-any-close"), pytest.param("fetch_open_orders", 2, id="final-orders"), - pytest.param("fetch_positions", 2, id="final-positions"), + # The second positions read is the one after the close. + pytest.param("fetch_positions", 3, id="final-positions"), ], ) def test_a_failed_venue_read_stops_flatten_and_never_reads_as_flat( @@ -417,9 +697,148 @@ def test_a_lost_close_is_reconciled_to_a_verdict_and_the_run_stops_with_the_size assert close.state is OrderState.FAILED +class _LostOverANetwork(_DropsEveryPlace): + """Every send is lost, and each order read waits on the network. + + ``places`` lists the symbol of each send in order. ``overlap`` is the most + order reads that were in flight at once. A pass reads its orders one by one, + so more than one means two passes overlapped. + """ + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self.places: list[str] = [] + self.overlap = 0 + self._in_flight = 0 + + async def place(self, order: PlaceOrder) -> None: + self.places.append(order.symbol) + + async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: + self._in_flight += 1 + self.overlap = max(self.overlap, self._in_flight) + # The network round trip, where another loop gets to run. + await asyncio.sleep(0) + self._in_flight -= 1 + return await super().fetch_order(ref) + + +def test_two_held_symbols_retry_at_the_same_time_and_their_passes_never_overlap( + tmp_path: Path, +) -> None: + # One loop per symbol, all at the same time, and one reconcile pass at a time + # under a lock. A loop that needs a pass waits for the one in progress. + _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) + venues: list[_LostOverANetwork] = [] + + def link(paper: PaperExchange) -> Exchange: + venues.append(_LostOverANetwork(paper)) + return venues[0] + + exit_code, logs = _flatten(tmp_path, link=link) + + assert exit_code == 1 + [venue] = venues + # Both symbols send a first close before either sends a second one. + assert sorted(venue.places[:2]) == ["BTC", "ETH"] + assert sorted(venue.places) == ["BTC"] * 3 + ["ETH"] * 3 + assert venue.overlap == 1 + gave_up = [log["symbol"] for log in logs if log["event"] == "flatten.gave_up"] + assert sorted(gave_up) == ["BTC", "ETH"] + + +class _PlacesOverANetwork(VenueLink): + """Each send waits on the network before the venue takes it, as an HTTP send does. + + ``places`` lists the symbol of each send in order. + """ + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self.places: list[str] = [] + + async def place(self, order: PlaceOrder) -> None: + self.places.append(order.symbol) + # The network round trip, where another loop gets to run. + await asyncio.sleep(0) + await super().place(order) + + +def test_a_send_that_waits_on_the_network_gets_no_second_close_beside_it( + tmp_path: Path, +) -> None: + # While one loop's send waits, another loop may run. Its close must still be + # handled before it looks for a verdict. Otherwise it reads no fill and sends + # a second close beside the first (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) + venues: list[_PlacesOverANetwork] = [] + + def link(paper: PaperExchange) -> Exchange: + venues.append(_PlacesOverANetwork(paper)) + return venues[0] + + exit_code, logs = _flatten(tmp_path, link=link) + + assert exit_code == 0 + [venue] = venues + assert sorted(venue.places) == ["BTC", "ETH"] + assert not [log for log in logs if log["event"].startswith("flatten.")] + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("trivial", "ETH"): (Decimal("0"), Decimal("100")), + } + + +class _NeverHandlesCloses: + """An ``InMemoryBus`` that takes every ``__operator__`` close and never handles it. + + ``closes`` counts them. This is a publish that returned before the engine handled it. + """ + + def __init__(self) -> None: + self._bus = InMemoryBus() + self.closes = 0 + + def subscribe[E: Event](self, event_type: type[E], handler: Handler[E]) -> None: + self._bus.subscribe(event_type, handler) + + async def start(self) -> None: + await self._bus.start() + + async def publish(self, event: Event) -> None: + if isinstance(event, PlaceSignal) and event.strategy_id == OPERATOR: + self.closes += 1 + return + await self._bus.publish(event) + + async def drain(self) -> None: + await self._bus.drain() + + async def close(self) -> None: + await self._bus.close() + + +def test_a_close_the_engine_never_handled_stops_the_run_with_no_second_close( + tmp_path: Path, +) -> None: + # With no saga there is no verdict to read. Reading it as a dry attempt would + # send a second close, and the first could still reach the venue (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + bus = _NeverHandlesCloses() + + exit_code, logs = _flatten(tmp_path, bus=bus) + + assert exit_code == 1 + assert bus.closes == 1 + assert not [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [log for log in logs if log["event"] == "engine.faulted"] + assert logs[-1]["outcome"] == "stopped" + + def test_a_second_flatten_closes_with_a_seq_above_the_first_ones_close(tmp_path: Path) -> None: # The manager drops a signal id it has seen (ADR-0006). So a second flatten that - # reused seq 1 would send nothing, and the size would stay open. + # reused seq 1 would send nothing, and the size would stay open. The first one + # spends seqs 1 to 3 on its three lost closes. _hold(tmp_path / "saga.db", {"BTC": "0.02"}) assert _flatten(tmp_path, link=_DropsEveryPlace)[0] == 1 @@ -429,7 +848,7 @@ def test_a_second_flatten_closes_with_a_seq_above_the_first_ones_close(tmp_path: assert logs[-1]["outcome"] == "done" assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} with SQLiteStore(tmp_path / "saga.db") as store: - close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:2")) + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:4")) assert close is not None assert close.state is OrderState.FILLED @@ -448,19 +867,84 @@ async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: return VenueReadFailure.SEND_FAILED -def test_a_failed_reconcile_pass_ends_the_wait_and_the_final_read_decides( - tmp_path: Path, -) -> None: - # The saga stays open for the next boot to resume (#474). The run never hangs. +def test_three_failed_reconcile_passes_give_up_with_no_second_close(tmp_path: Path) -> None: + # A failed read is a dry attempt (ADR-0060), so the run never hangs. The close + # could still fill, so no second one goes beside it. The saga stays open for + # the next boot to resume (#474). _hold(tmp_path / "saga.db", {"BTC": "0.02"}) exit_code, logs = _flatten(tmp_path, link=_LostAndUnreadable) assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [ + (1, "read_failed"), + (2, "read_failed"), + (3, "read_failed"), + ] + assert [log["symbol"] for log in logs if log["event"] == "flatten.gave_up"] == ["BTC"] + assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] assert logs[-1]["outcome"] == "stopped" assert logs[-1]["left"] == "BTC=0.02" +def test_a_failed_read_after_a_close_is_a_dry_attempt_and_the_next_read_decides( + tmp_path: Path, +) -> None: + # The close filled, but the read that would say so failed. Flatten waits and + # reads again instead of stopping (ADR-0060). It sends no second close. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + + # The first positions read is flatten's own. The second is the one after the close. + exit_code, logs = _flatten(tmp_path, link=partial(_FailsOneRead, read="fetch_positions", nth=2)) + + assert exit_code == 0 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [(1, "read_failed")] + assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] + assert logs[-1]["outcome"] == "done" + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + + +class _ReadsPositionsOnce(VenueLink): + """Only the first positions read answers. Every later one fails.""" + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self._reads = 0 + + async def fetch_positions(self) -> dict[str, Decimal] | VenueReadFailure: + self._reads += 1 + if self._reads > 1: + return VenueReadFailure.SEND_FAILED + return await super().fetch_positions() + + +def test_a_venue_that_never_answers_after_a_close_gives_up_and_the_run_ends( + tmp_path: Path, +) -> None: + # Each failed read is a dry attempt, so a venue that cannot answer never hangs + # the run (ADR-0060). It sends no second close. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + + exit_code, logs = _flatten(tmp_path, link=_ReadsPositionsOnce) + + assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [ + (1, "read_failed"), + (2, "read_failed"), + (3, "read_failed"), + ] + # The close filled, but no read could say so. So the event names the last + # size the venue reported, from before the close. + gave_up = [log for log in logs if log["event"] == "flatten.gave_up"] + assert [(log["symbol"], log["size"]) for log in gave_up] == [("BTC", "0.02")] + assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] + assert logs[-1]["outcome"] == "stopped" + assert logs[-1]["left"] == f"read_failed: {VenueReadFailure.SEND_FAILED.value}" + + def _left_open(path: Path, symbol: str, *, quantity: str = "0.02") -> str: """Leave an ``__operator__`` close ``LIVE`` in the store, one paper never lists. @@ -527,33 +1011,47 @@ async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: def test_a_close_left_open_that_no_pass_can_decide_gets_no_second_close_beside_it( tmp_path: Path, ) -> None: - # The wait ends on a failed pass (ADR-0060), with the old close still open. - # The run stops with the size left, and the next boot resumes the saga. + # Each failed pass is a dry attempt, as it is after a close (ADR-0060). After 3 + # the old close is still open. The run stops with the size left, and the next + # boot resumes the saga. _hold(tmp_path / "saga.db", {"BTC": "0.02"}) _left_open(tmp_path / "saga.db", "BTC") exit_code, logs = _flatten(tmp_path, link=_ReadsOnlyAtBoot) assert exit_code == 1 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [ + (1, "read_failed"), + (2, "read_failed"), + (3, "read_failed"), + ] + assert [log["symbol"] for log in logs if log["event"] == "flatten.gave_up"] == ["BTC"] assert logs[-1]["left"] == "BTC=0.02" assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] -def test_a_failed_read_after_the_wait_stops_flatten_with_no_second_close( +def test_a_failed_read_after_the_wait_is_a_dry_attempt_and_the_next_read_sizes_the_close( tmp_path: Path, ) -> None: - # The old close may have filled during the wait. With no read of what is - # left, flatten cannot size a close, so it sends none (ADR-0060). + # The old close may have filled during the wait. With no read of what is left, + # flatten cannot size a close (ADR-0060). So it waits and reads again, as it + # does after a close, and only that read sizes the next one. _hold(tmp_path / "saga.db", {"BTC": "0.02"}) _left_open(tmp_path / "saga.db", "BTC") # The first positions read is flatten's own. The second is the read after the wait. exit_code, logs = _flatten(tmp_path, link=partial(_FailsOneRead, read="fetch_positions", nth=2)) - assert exit_code == 1 - assert logs[-1]["outcome"] == "stopped" - assert logs[-1]["left"] == f"read_failed: {VenueReadFailure.SEND_FAILED.value}" - assert not [log for log in logs if log.get("signal_id") == f"{OPERATOR}:BTC:2"] + assert exit_code == 0 + dry = [log for log in logs if log["event"] == "flatten.dry_attempt"] + assert [(log["count"], log["reason"]) for log in dry] == [(1, "read_failed")] + events = [(log["event"], log.get("signal_id")) for log in logs] + assert events.index(("flatten.dry_attempt", None)) < events.index( + ("order.placed", f"{OPERATOR}:BTC:2") + ) + assert logs[-1]["outcome"] == "done" + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} class _Venue: diff --git a/tests/observability/test_catalog_walk.py b/tests/observability/test_catalog_walk.py index f4ebfa98..7219e45c 100644 --- a/tests/observability/test_catalog_walk.py +++ b/tests/observability/test_catalog_walk.py @@ -797,6 +797,28 @@ def _drive_exit_unguarded() -> None: assert asyncio.run(engine.run()) == 0 +def _drive_flatten_gave_up() -> None: + """One flatten of a position the idle feed never prices, so every attempt is dry (ADR-0056).""" + bus = InMemoryBus() + clock = ManualClock() + engine = Engine( + bus=bus, + clock=clock, + store=SQLiteStore(":memory:"), + exchange=PaperExchange( + bus=bus, + clock=clock, + fill_model=ImmediateFillModel(), + genesis_collateral=GENESIS, + account_net=lambda: {"BTC": Decimal("0.02")}, + applied_fills=lambda cloid: (), + ), + feed=_IdleFeed(), + exit_job=Flatten(), + ) + assert asyncio.run(engine.run()) == 1 + + def _drive_exit_refused() -> None: """One operator cancel all on a store another engine holds (ADR-0060).""" with tempfile.TemporaryDirectory() as tmp, SQLiteStore(Path(tmp) / "saga.db") as holder: @@ -1125,6 +1147,8 @@ async def go() -> None: NamedEvent.EXIT_FINISHED: _drive_exit_finished, NamedEvent.EXIT_UNGUARDED: _drive_exit_unguarded, NamedEvent.CANCEL_ALL_ORDERS_REMAIN: _drive_cancel_all_orders_remain, + NamedEvent.FLATTEN_DRY_ATTEMPT: _drive_flatten_gave_up, + NamedEvent.FLATTEN_GAVE_UP: _drive_flatten_gave_up, NamedEvent.GUARD_KILL_SWITCH_TRIPPED: _drive_kill_switch(reset=False), NamedEvent.GUARD_KILL_SWITCH_RESET: _drive_kill_switch(reset=True), NamedEvent.STRATEGY_ERROR: _drive_strategy_error,