diff --git a/CONTEXT.md b/CONTEXT.md index 8ef6def5..073c4c38 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -477,8 +477,8 @@ partition at the strategy's entry price. A failed venue read also counts as an a nothing (ADR-0060). A read refused for a rate limit does not count. It waits like an order. Flatten is done only when the venue holds no resting order and no position. The fill split is built (#471). #472 built one attempt per symbol on paper replay. A symbol the feed never priced gets no -order, and the final read reports it. The size check before an order moved to #475. The rest is -not built yet (#408). +order, and the final read reports it. The size check before an order moved to #475. #474 built +the resume after a crash. Each close takes a new seq. The rest is not built yet (#408). _Avoid_: close out, liquidate (that is the venue's forced close), panic sell. **Reduce-only order**: diff --git a/README.md b/README.md index eb8be738..a52402f0 100644 --- a/README.md +++ b/README.md @@ -191,9 +191,6 @@ A position the feed never priced is not closed. The run exits `1`, and `exit.fin symbol and its size, for example `left=BTC=0.02`. Under `GUARD=noop` there is no kill switch to trip, so the run warns with `exit.unguarded`. -Flatten closes each symbol only once per store today. If a later run opens that symbol again, the -next flatten sends no close for it. It exits `1` and names the symbol in `left`. #474 fixes this. - ## What it does today - Venues: [Hyperliquid](https://hyperliquid.gitbook.io/hyperliquid-docs) (mainnet and testnet), diff --git a/docs/module-maps/safe-exit.md b/docs/module-maps/safe-exit.md index 7be3453f..42c0b4d6 100644 --- a/docs/module-maps/safe-exit.md +++ b/docs/module-maps/safe-exit.md @@ -162,12 +162,18 @@ context. Starts the feed through the `FeedHandle`. - **(Deferred by #472:** every close uses seq 1. The manager drops a signal id it has seen (ADR-0006). So a second flatten on the same store sends no close for a symbol an earlier 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.**)** **The per-symbol loop (private):** - Before the first attempt, wait for the verdict on any open `__operator__` saga for the symbol that the boot left open. It waits the same way as after an attempt, below. A symbol never has two attempts open. +- **(Built by #474, before the loops of #473:** the wait runs per symbol, after the feed has + 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.**)** - 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()`. diff --git a/src/tickwright/engine/cache.py b/src/tickwright/engine/cache.py index c065d4f9..546c2725 100644 --- a/src/tickwright/engine/cache.py +++ b/src/tickwright/engine/cache.py @@ -7,7 +7,7 @@ checkpoint path. """ -from tickwright.domain import Order, Store +from tickwright.domain import Order, SignalId, Store class Cache: @@ -70,6 +70,25 @@ def all_orders(self) -> list[Order]: """ return list(self._orders.values()) + def seq_high_water(self) -> dict[str, int]: + """Max consumed seq per strategy across every saga (ADR-0016). + + Every place consumed its ``signal_id``'s seq, and every requested + cancel consumed its own (ADR-0026 — omitting cancels would let a + restart reuse a consumed id). One per-strategy counter across all + symbols, so this is a single max over the strategy's records. The + strategies and ``Flatten`` both start their counters here, so neither + can reuse an id after a restart. + """ + high_water: dict[str, int] = {} + for order in self.all_orders(): + seqs = [SignalId.parse(order.signal_id).seq] + if order.cancel_signal_id is not None: + seqs.append(SignalId.parse(order.cancel_signal_id).seq) + current = high_water.get(order.strategy_id, 0) + high_water[order.strategy_id] = max(current, *seqs) + return high_water + def last_event_ts(self, cloid: str) -> int | None: """The ``ts_ns`` of ``cloid``'s most recent checkpoint *this session*, or ``None`` if it has not been written since the last ``rebuild`` — the diff --git a/src/tickwright/engine/exit_run.py b/src/tickwright/engine/exit_run.py index 8ab35099..73faa6fe 100644 --- a/src/tickwright/engine/exit_run.py +++ b/src/tickwright/engine/exit_run.py @@ -138,15 +138,48 @@ 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 in priced: - await _await_verdict(await _close(symbol, size, context), context) + 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) return await _final_read(context) -async def _close(symbol: str, size: Decimal, context: ExitContext) -> str: +async def _resume_closes( + symbol: str, size: Decimal, context: ExitContext +) -> Decimal | VenueReadFailure: + """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. + """ + 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)) + + +async def _close(symbol: str, size: Decimal, seq: int, context: ExitContext) -> str: """Send one reduce-only market IOC order for the whole of ``size`` (ADR-0054). Returns the cloid of the order's saga. @@ -157,7 +190,7 @@ async def _close(symbol: str, size: Decimal, context: ExitContext) -> str: ts_init=now, strategy_id=OPERATOR, symbol=symbol, - seq=1, + seq=seq, side=Side.SELL if size > 0 else Side.BUY, quantity=abs(size), order_type=OrderType.MARKET, diff --git a/src/tickwright/engine/strategy_host.py b/src/tickwright/engine/strategy_host.py index fecaeb32..bf6c177a 100644 --- a/src/tickwright/engine/strategy_host.py +++ b/src/tickwright/engine/strategy_host.py @@ -20,7 +20,6 @@ InvariantViolation, MarketTick, OrderEvent, - SignalId, Store, Strategy, SymbolOwnership, @@ -119,7 +118,7 @@ def start(self) -> None: ``Cache`` the runner rebuilt from that store, so the boot deserializes the history once (issue #233). """ - high_water = self._seq_high_water() + high_water = self._cache.seq_high_water() for strategy in self._strategies.values(): self._restore(strategy) # Probe once before any tick: a strategy the host cannot persist @@ -129,23 +128,6 @@ def start(self) -> None: self._subscribe(strategy) self._started = True - def _seq_high_water(self) -> dict[str, int]: - """Max consumed seq per strategy across every checkpointed saga. - - Every place consumed its ``signal_id``'s seq, and every requested - cancel consumed its own (ADR-0026 — omitting cancels would let a - restart reuse a consumed id). One per-strategy counter across all - symbols, so this is a single max over the strategy's records. - """ - high_water: dict[str, int] = {} - for order in self._cache.all_orders(): - seqs = [SignalId.parse(order.signal_id).seq] - if order.cancel_signal_id is not None: - seqs.append(SignalId.parse(order.cancel_signal_id).seq) - current = high_water.get(order.strategy_id, 0) - high_water[order.strategy_id] = max(current, *seqs) - return high_water - def _restore(self, strategy: Strategy) -> None: """Feed the persisted snapshot to ``strategy.restore()``, if one exists. diff --git a/tests/engine/test_flatten_e2e.py b/tests/engine/test_flatten_e2e.py index 072f73cd..4c1a9f44 100644 --- a/tests/engine/test_flatten_e2e.py +++ b/tests/engine/test_flatten_e2e.py @@ -47,7 +47,7 @@ ) from tickwright.engine.checkpoint import Checkpointer from tickwright.engine.execution import ExecutionManager -from tickwright.engine.exit_run import Flatten +from tickwright.engine.exit_run import ExitJob, Flatten, OperatorCancelAll from tickwright.engine.guard import NoopGuard, RealGuard from tickwright.engine.runner import Engine from tickwright.observability.testing import capture_events @@ -92,8 +92,8 @@ def _paper(bus: InMemoryBus, clock: ManualClock, store: SQLiteStore) -> PaperExc ) -def _hold(path: Path, sizes: dict[str, str]) -> None: - """Open each signed size as the strategy ``trivial``, at the symbol's first row price. +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. Seeded through the engine's own order path on a separate bus, as an earlier run would. """ @@ -114,7 +114,7 @@ async def go() -> None: PlaceSignal( ts_event=0, ts_init=0, - strategy_id="trivial", + strategy_id=strategy_id, symbol=symbol, seq=seq, side=Side.BUY if Decimal(size) > 0 else Side.SELL, @@ -417,6 +417,23 @@ def test_a_lost_close_is_reconciled_to_a_verdict_and_the_run_stops_with_the_size assert close.state is OrderState.FAILED +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. + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + assert _flatten(tmp_path, link=_DropsEveryPlace)[0] == 1 + + exit_code, logs = _flatten(tmp_path) + + assert exit_code == 0 + 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")) + assert close is not None + assert close.state is OrderState.FILLED + + class _LostAndUnreadable(_DropsEveryPlace): """The send is lost and every order read fails, so no pass can ever decide it.""" @@ -442,3 +459,323 @@ def test_a_failed_reconcile_pass_ends_the_wait_and_the_final_read_decides( assert exit_code == 1 assert logs[-1]["outcome"] == "stopped" assert logs[-1]["left"] == "BTC=0.02" + + +def _left_open(path: Path, symbol: str, *, quantity: str = "0.02") -> str: + """Leave an ``__operator__`` close ``LIVE`` in the store, one paper never lists. + + That is a close the venue acked and then dropped (#242) before a crash. The + boot can only arm its ghost window, so the saga is still open when flatten + starts. Returns its cloid. + """ + cloid = derive_cloid(f"{OPERATOR}:{symbol}:1") + clock = ManualClock() + with SQLiteStore(path) as store: + paper = _paper(InMemoryBus(), clock, store) + checks = Checkpointer(spec=paper.account_spec(), store=store, clock=clock, specs=_SPECS) + checks.recover() + checks.checkpoint( + Order( + cloid=cloid, + strategy_id=OPERATOR, + signal_id=f"{OPERATOR}:{symbol}:1", + symbol=symbol, + side=Side.SELL, + quantity=Decimal(quantity), + order_type=OrderType.MARKET, + state=OrderState.LIVE, + created_ts_ns=0, + # As the manager stores it at place, so a fill can be booked (ADR-0054). + split_basis=checks.split_basis(symbol), + ) + ) + return cloid + + +def test_flatten_waits_for_the_verdict_on_a_close_the_boot_left_open_before_it_closes_again( + tmp_path: Path, +) -> None: + # The dropped close could still have filled. A second close sent beside it + # could close the size twice, so a symbol never has two closes open (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + left_open = _left_open(tmp_path / "saga.db", "BTC") + + exit_code, logs = _flatten(tmp_path) + + assert exit_code == 0 + events = [(log["event"], log.get("cloid"), log.get("signal_id")) for log in logs] + verdict = events.index(("ghost.reconciled", left_open, None)) + placed = events.index(("order.placed", None, f"{OPERATOR}:BTC:2")) + assert verdict < placed + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + + +class _ReadsOnlyAtBoot(VenueLink): + """The boot reads every order. After that, no order read reaches the venue.""" + + def __init__(self, venue: Exchange) -> None: + super().__init__(venue) + self.reads = 0 + + async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: + self.reads += 1 + if self.reads == 1: + return await super().fetch_order(ref) + return VenueReadFailure.SEND_FAILED + + +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. + _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 + 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( + 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). + _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"] + + +class _Venue: + """Paper on a bus of its own, so it outlives the engine run it serves. + + The real venue keeps an order a crashed engine never booked. Paper keeps its + book in memory only (#484), so the test holds one instance across both runs. + """ + + def __init__(self, path: Path, clock: ManualClock) -> None: + self.bus = InMemoryBus() + self.store = SQLiteStore(path) + self._engine_bus: InMemoryBus | None = None + self.paper = PaperExchange( + bus=self.bus, + clock=clock, + fill_model=ImmediateFillModel(), + genesis_collateral=GENESIS, + instrument_specs=_SPECS, + account_net=lambda: paper_store_reads(self.store).account_net(), + applied_fills=lambda cloid: paper_store_reads(self.store).applied_fills(cloid), + ) + self.bus.subscribe(ExecutionReport, self._report) + + def serve(self, engine_bus: InMemoryBus) -> None: + """Link the venue to a new engine run: its ticks in, the venue's reports out.""" + self._engine_bus = engine_bus + + async def tick(tick: MarketTick) -> None: + await self.bus.publish(tick) + + engine_bus.subscribe(MarketTick, tick) + + def cut(self) -> None: + """The engine dies. Nothing the venue says from here reaches it.""" + self._engine_bus = None + + async def _report(self, report: ExecutionReport) -> None: + if self._engine_bus is not None: + await self._engine_bus.publish(report) + + +class _CrashesAfterTheSend(VenueLink): + """The close reaches the venue, and the engine dies before it hears the fill.""" + + def __init__(self, venue: Exchange, crash: Callable[[], None]) -> None: + super().__init__(venue) + self._crash = crash + self.crashed = asyncio.Event() + + async def place(self, order: PlaceOrder) -> None: + self._crash() + await super().place(order) + self.crashed.set() + await asyncio.Event().wait() + + +def _engine( + venue: _Venue, exchange: Exchange, tmp_path: Path, job: ExitJob | None = None +) -> Engine: + """One engine run of ``job``, flatten by default, served by ``venue`` through ``exchange``. + + Each run replays the file from its first row, so each gets its own clock. + """ + bus = InMemoryBus() + clock = ManualClock() + venue.serve(bus) + return Engine( + bus=bus, + clock=clock, + store=venue.store, + exchange=exchange, + feed=ReplayFeed(path=_ticks(tmp_path / "ticks.jsonl", _ROWS), bus=bus, clock=clock), + guard=RealGuard(specs=_SPECS, store=venue.store, clock=clock), + exit_job=Flatten() if job is None else job, + ) + + +def _crash_a_flatten(tmp_path: Path) -> _Venue: + """Run flatten until its first close reaches paper, then kill the run. + + Returns the venue that still holds the fill, with the store open again. + """ + venue = _Venue(tmp_path / "saga.db", ManualClock()) + link = _CrashesAfterTheSend(venue.paper, venue.cut) + + async def first_life() -> None: + run = asyncio.create_task(_engine(venue, link, tmp_path).run()) + await asyncio.wait_for(link.crashed.wait(), timeout=5) + run.cancel() + await asyncio.gather(run, return_exceptions=True) + + asyncio.run(first_life()) + venue.store.close() + venue.store = SQLiteStore(tmp_path / "saga.db") + return venue + + +def _next_run( + venue: _Venue, + tmp_path: Path, + job: ExitJob, + *, + link: Callable[[PaperExchange], Exchange] | None = None, +) -> tuple[int, list[EventDict]]: + """Run ``job`` after the crash. Returns the exit code and every event it emitted. + + ``link`` puts a venue double in front of paper. + """ + exchange = venue.paper if link is None else link(venue.paper) + + async def go() -> int: + return await asyncio.wait_for(_engine(venue, exchange, tmp_path, job).run(), 5) + + with capture_events() as logs: + exit_code = asyncio.run(go()) + venue.store.close() + return exit_code, logs + + +def test_a_flatten_that_crashed_after_its_send_is_booked_by_the_split_and_the_rest_closed( + tmp_path: Path, +) -> None: + # Two partitions share BTC, so the boot books the resumed fill by its stored + # split (ADR-0054). The other symbol is still open, and the next run closes it. + _hold(tmp_path / "saga.db", {"BTC": "0.02", "ETH": "0.5"}) + _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") + venue = _crash_a_flatten(tmp_path) + + exit_code, logs = _next_run(venue, tmp_path, Flatten()) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + # Opened at 42000 and 3000. Closed at the last rows, 44000 and 3200 (ADR-0059). + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("other", "BTC"): (Decimal("0"), Decimal("20")), + ("trivial", "ETH"): (Decimal("0"), Decimal("100")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + closes = [order for order in store.all_orders() if order.strategy_id == OPERATOR] + # One close per symbol, so no size was closed twice. The resumed run's seq is + # above the crashed run's. + assert sorted((order.symbol, order.signal_id) for order in closes) in ( + [("BTC", f"{OPERATOR}:BTC:1"), ("ETH", f"{OPERATOR}:ETH:2")], + [("BTC", f"{OPERATOR}:BTC:2"), ("ETH", f"{OPERATOR}:ETH:1")], + ) + assert all(order.state is OrderState.FILLED for order in closes) + + +def test_a_cancel_all_after_a_crashed_flatten_books_the_close_at_boot(tmp_path: Path) -> None: + # The resume is the boot, not the flatten job, so any exit run does it (ADR-0056). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + _hold(tmp_path / "saga.db", {"BTC": "0.01"}, strategy_id="other") + venue = _crash_a_flatten(tmp_path) + + exit_code, logs = _next_run(venue, tmp_path, OperatorCancelAll()) + + assert exit_code == 0 + assert logs[-1]["job"] == "cancel_all" + assert _partitions(tmp_path) == { + ("trivial", "BTC"): (Decimal("0"), Decimal("40")), + ("other", "BTC"): (Decimal("0"), Decimal("20")), + } + with SQLiteStore(tmp_path / "saga.db") as store: + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:1")) + assert close is not None + assert close.state is OrderState.FILLED + + +class _FillsDuringTheWait(VenueLink): + """``close`` is unknown to the boot's read. It fills just before the next read. + + So it fills after flatten's first position read and before the wait ends. + """ + + def __init__(self, venue: Exchange, *, close: PlaceOrder) -> None: + super().__init__(venue) + self._close = close + self._reads = 0 + + async def fetch_order(self, ref: OrderRef) -> VenueOrderView | VenueReadFailure: + if ref.cloid == self._close.cloid: + self._reads += 1 + if self._reads == 1: + return VenueOrderView(status=None) + if self._reads == 2: + await self.place(self._close) + return await super().fetch_order(ref) + + +def test_a_close_left_open_that_fills_during_the_wait_leaves_only_the_rest_to_close( + tmp_path: Path, +) -> None: + # A close sized from the read before the wait would ask for the whole size + # again. Paper would shrink it to what is left and cancel the rest (ADR-0057). + _hold(tmp_path / "saga.db", {"BTC": "0.02"}) + left_open = _left_open(tmp_path / "saga.db", "BTC", quantity="0.01") + late = PlaceOrder( + cloid=left_open, + symbol="BTC", + side=Side.SELL, + quantity=Decimal("0.01"), + order_type=OrderType.MARKET, + time_in_force=TimeInForce.IOC, + reduce_only=True, + ) + venue = _Venue(tmp_path / "saga.db", ManualClock()) + + exit_code, logs = _next_run( + venue, tmp_path, Flatten(), link=partial(_FillsDuringTheWait, close=late) + ) + + assert exit_code == 0 + assert logs[-1]["outcome"] == "done" + # Opened at 42000. Both halves close at the last row, 44000 (ADR-0059). + assert _partitions(tmp_path) == {("trivial", "BTC"): (Decimal("0"), Decimal("40"))} + with SQLiteStore(tmp_path / "saga.db") as store: + old = store.get_order(left_open) + close = store.get_order(derive_cloid(f"{OPERATOR}:BTC:2")) + assert old is not None + assert old.state is OrderState.FILLED + assert close is not None + assert close.quantity == Decimal("0.01") + assert close.state is OrderState.FILLED